Repository: cloudnative-pg/cloudnative-pg

Issue:
[Bug]: Replication slots not cleaned up on replicas when both HA and synchronizeReplicas are disabled
### Is there an existing issue already for this bug?

- [x] I have searched for an existing issue, and could not find anything. I believe this is a new bug.

### I have read the troubleshooting guide

- [x] I have read the troubleshooting guide and I think this is a new bug.

### I am running a supported version of CloudNativePG

- [x] I have read the troubleshooting guide and I think this is a new bug.

### Contact Details

_No response_

### Version

trunk (main)

### What version of Kubernetes are you using?

1.34

### What is your Kubernetes environment?

Self-managed: kind (evaluation)

### How did you install the operator?

YAML manifest

### What happened?

User-defined replication slots are not cleaned up from replica pods when both `replicationSlots.highAvailability.enabled` and `replicationSlots.synchronizeReplicas.enabled` are disabled.

This is the root cause of the flaky E2E test in #8968, which was tentatively addressed in #8971 by increasing the timeout; however, the actual bug was not resolved.

### Root Cause

When replication slots are disabled, the `Replicator` stops without performing cleanup:

```go
if config == nil || !config.GetEnabled() {
    ticker.Stop()
    updateInterval = 0
    continue  // ← BUG: No cleanup happens
}
```

When both features are disabled, `config.GetEnabled()` returns `false`, causing the loop to continue without calling `sr.reconcile()`. The cleanup logic in `synchronizeReplicationSlots()` is never executed.



### Cluster resource

```shell

```

### Relevant log output

```shell
No errors logged - the Replicator silently stops without cleanup.
```

### Code of Conduct

- [x] I agree to follow this project's Code of Conduct

Discussion:
@dosubot: <!-- Answer -->
This is a new, previously undocumented bug: when both `replicationSlots.highAvailability.enabled` and `replicationSlots.synchronizeReplicas.enabled` are disabled, the Replicator process stops without performing cleanup, so user-defined replication slots are not removed from replicas. This is confirmed by the code in `internal/management/controller/slots/runner/runner.go`, where the main loop exits early and never calls the cleanup logic in `synchronizeReplicationSlots` if both features are disabled. As a result, slots linger on replicas, which explains the flaky E2E test in #8968. The recent PR #8971 only increased test timeouts and did not address the root cause—no merged PRs or issues currently fix this behavior. The expected cleanup (removal of slots when synchronization is disabled) is documented, but the implementation does not trigger it in this scenario, so manual cleanup or a code change is needed to resolve the issue [source](https://github.com/cloudnative-pg/cloudnative-pg/blob/cacd7498f85b62fdabf1971d2d855a48831d973c/internal/management/controller/slots/runner/runner.go#L67-L219) [docs](https://cloudnative-pg.io/documentation/current/replication/).


<!-- Dosu Comment Footer -->
*To reply, just mention [@dosu](https://go.dosu.dev/dosubot).*

---
<sup>How did I do? [Good](https://app.dosu.dev/response-feedback/a1a624b8-a12b-4f93-8868-516e5655129e?feedback_type=great_response) | [Irrelevant](https://app.dosu.dev/response-feedback/a1a624b8-a12b-4f93-8868-516e5655129e?feedback_type=irrelevant_answer) | [Incorrect](https://app.dosu.dev/response-feedback/a1a624b8-a12b-4f93-8868-516e5655129e?feedback_type=incorrect_sources) | [Verbose](https://app.dosu.dev/response-feedback/a1a624b8-a12b-4f93-8868-516e5655129e?feedback_type=too_verbose) | [Hallucination](https://app.dosu.dev/response-feedback/a1a624b8-a12b-4f93-8868-516e5655129e?feedback_type=hallucination) | [Report 🐛](https://app.dosu.dev/response-feedback/a1a624b8-a12b-4f93-8868-516e5655129e?feedback_type=bug_report) | [Other](https://app.dosu.dev/response-feedback/a1a624b8-a12b-4f93-8868-516e5655129e?feedback_type=other)</sup>

[![Chat with Dosu](https://dosu.dev/dosu-chat-badge.svg)](https://app.dosu.dev/89b818f3-8b5e-449b-bfb3-64a9360d9665/ask?utm_source=github)&nbsp;[![Open in Cursor](https://dosu.dev/dosu-in-cursor.svg)](https://cursor.com/link/prompt?text=This%20is%20a%20new%2C%20previously%20undocumented%20bug%3A%20when%20both%20%60replicationSlots.highAvailability.enabled%60%20and%20%60replicationSlots.synchronizeReplicas.enabled%60%20are%20disabled%2C%20the%20Replicator%20process%20stops%20without%20performing%20cleanup%2C%20so%20user-defined%20replication%20slots%20are%20not%20removed%20from%20replicas.%20This%20is%20confirmed%20by%20the%20code%20in%20%60internal/management/controller/slots/runner/runner.go%60%2C%20where%20the%20main%20loop%20exits%20early%20and%20never%20calls%20the%20cleanup%20logic%20in%20%60synchronizeReplicationSlots%60%20if%20both%20features%20are%20disabled.%20As%20a%20result%2C%20slots%20linger%20on%20replicas%2C%20which%20explains%20the%20flaky%20E2E%20test%20in%20%238968.%20The%20recent%20PR%20%238971%20only%20increased%20test%20timeouts%20and%20did%20not%20address%20the%20root%20cause%E2%80%94no%20merged%20PRs%20or%20issues%20currently%20fix%20this%20behavior.%20The%20expected%20cleanup%20%28removal%20of%20slots%20when%20synchronization%20is%20disabled%29%20is%20documented%2C%20but%20the%20implementation%20does%20not%20trigger%20it%20in%20this%20scenario%2C%20so%20manual%20cleanup%20or%20a%20code%20change%20is%20needed%20to%20resolve%20the%20issue%20%5Bsource%5D%28https%3A//github.com/cloudnative-pg/cloudnative-pg/blob/cacd7498f85b62fdabf1971d2d855a48831d973c/internal/management/controller/slots/runner/runner.go%23L67-L219%29%20%5Bdocs%5D%28https%3A//cloudnative-pg.io/documentation/current/replication/%29.)&nbsp;[![Join Discord](https://img.shields.io/badge/join-5865F2?logo=discord&logoColor=white&label=)](https://go.dosu.dev/discord-bot)&nbsp;[![Share on X](https://img.shields.io/badge/X-share-black)](https://twitter.com/intent/tweet?text=%40dosu_ai%20helped%20me%20solve%20this%20issue!&url=https%3A//github.com/cloudnative-pg/cloudnative-pg/issues/9380)

Solution patch:
diff --git a/internal/management/controller/slots/runner/runner.go b/internal/management/controller/slots/runner/runner.go
index e3bd9e12..9aff4160 100644
--- a/internal/management/controller/slots/runner/runner.go
+++ b/internal/management/controller/slots/runner/runner.go
@@ -72,13 +72,18 @@ func (sr *Replicator) Start(ctx context.Context) error {
 			case <-ticker.C:
 			}
 
-			// If replication is disabled stop the timer,
-			// the process will resume through the wakeUp channel if necessary
+			// If replication is disabled, stop the timer and perform cleanup
 			if config == nil || !config.GetEnabled() {
 				ticker.Stop()
-				// we set updateInterval to 0 to make sure the Ticker will be reset
+				// We set updateInterval to 0 to make sure the Ticker will be reset
 				// if the feature is enabled again
 				updateInterval = 0
+				// Perform cleanup before continuing
+				if config != nil {
+					if err := sr.reconcile(ctx, config); err != nil {
+						contextLog.Warning("cleanup during reconcile when features are disabled", "err", err)
+					}
+				}
 				continue
 			}
 
diff --git a/internal/webhook/v1/cluster_webhook.go b/internal/webhook/v1/cluster_webhook.go
index a59033bd..8d90c6b1 100644
--- a/internal/webhook/v1/cluster_webhook.go
+++ b/internal/webhook/v1/cluster_webhook.go
@@ -2273,33 +2273,54 @@ func (v *ClusterCustomValidator) validateReplicationSlotsChange(r, old *apiv1.Cl
 	newReplicationSlots := r.Spec.ReplicationSlots
 	oldReplicationSlots := old.Spec.ReplicationSlots
 
-	if oldReplicationSlots == nil || oldReplicationSlots.HighAvailability == nil ||
-		!oldReplicationSlots.HighAvailability.GetEnabled() {
+	var errs field.ErrorList
+
+	if oldReplicationSlots == nil {
 		return nil
 	}
 
-	var errs field.ErrorList
+	// Validate HighAvailability changes
+	if oldReplicationSlots.HighAvailability.GetEnabled() {
+		// When disabling, we should check that the prefix is not removed and doesn't change
+		// to properly execute the cleanup logic
+		if newReplicationSlots == nil || newReplicationSlots.HighAvailability == nil {
+			path := field.NewPath("spec", "replicationSlots")
+			if newReplicationSlots != nil {
+				path = path.Child("highAvailability")
+			}
+			errs = append(errs,
+				field.Invalid(
+					path,
+					nil,
+					fmt.Sprintf("Cannot remove %v section while highAvailability is enabled", path)),
+			)
+		} else if oldReplicationSlots.HighAvailability.SlotPrefix != newReplicationSlots.HighAvailability.SlotPrefix {
+			errs = append(errs,
+				field.Invalid(
+					field.NewPath("spec", "replicationSlots", "highAvailability", "slotPrefix"),
+					newReplicationSlots.HighAvailability.SlotPrefix,
+					"Cannot change replication slot prefix while highAvailability is enabled"),
+			)
+		}
+	}
 
-	// when disabling we should check that the prefix it's not removed, and it doesn't change to
-	// properly execute the cleanup logic
-	if newReplicationSlots == nil || newReplicationSlots.HighAvailability == nil {
-		path := field.NewPath("spec", "replicationSlots")
-		if newReplicationSlots != nil {
-			path = path.Child("highAvailability")
+	// Validate SynchronizeReplicas changes
+	// When synchronizeReplicas is enabled, we need to ensure users disable it before removing the configuration
+	// to allow the cleanup logic to properly remove user-defined replication slots from replicas
+	if oldReplicationSlots.SynchronizeReplicas.GetEnabled() {
+		if newReplicationSlots == nil || newReplicationSlots.SynchronizeReplicas == nil {
+			path := field.NewPath("spec", "replicationSlots")
+			if newReplicationSlots != nil {
+				path = path.Child("synchronizeReplicas")
+			}
+			errs = append(errs,
+				field.Invalid(
+					path,
+					nil,
+					fmt.Sprintf("Cannot remove %v section while synchronizeReplicas is enabled. "+
+						"Disable synchronizeReplicas first to allow cleanup of user-defined replication slots on replicas", path)),
+			)
 		}
-		errs = append(errs,
-			field.Invalid(
-				path,
-				nil,
-				fmt.Sprintf("Cannot remove %v section while highAvailability is enabled", path)),
-		)
-	} else if oldReplicationSlots.HighAvailability.SlotPrefix != newReplicationSlots.HighAvailability.SlotPrefix {
-		errs = append(errs,
-			field.Invalid(
-				field.NewPath("spec", "replicationSlots", "highAvailability", "slotPrefix"),
-				newReplicationSlots.HighAvailability.SlotPrefix,
-				"Cannot change replication slot prefix while highAvailability is enabled"),
-		)
 	}
 
 	return errs



Test patch:
diff --git a/internal/webhook/v1/cluster_webhook_test.go b/internal/webhook/v1/cluster_webhook_test.go
index 57edbe3f..154d9032 100644
--- a/internal/webhook/v1/cluster_webhook_test.go
+++ b/internal/webhook/v1/cluster_webhook_test.go
@@ -3546,6 +3546,9 @@ var _ = Describe("validation of replication slots configuration", func() {
 					HighAvailability: &apiv1.ReplicationSlotsHAConfiguration{
 						Enabled: ptr.To(false),
 					},
+					SynchronizeReplicas: &apiv1.SynchronizeReplicasConfiguration{
+						Enabled: ptr.To(false),
+					},
 				},
 			},
 		}
@@ -3557,6 +3560,9 @@ var _ = Describe("validation of replication slots configuration", func() {
 				Enabled:    ptr.To(true),
 				SlotPrefix: "_test_",
 			},
+			SynchronizeReplicas: &apiv1.SynchronizeReplicasConfiguration{
+				Enabled: ptr.To(true),
+			},
 		}
 
 		Expect(v.validateReplicationSlotsChange(newCluster, oldCluster)).To(BeEmpty())
@@ -3581,7 +3587,7 @@ var _ = Describe("validation of replication slots configuration", func() {
 		Expect(v.validateReplicationSlotsChange(newCluster, oldCluster)).To(HaveLen(1))
 	})
 
-	It("prevents removing the replication slot section when replication slots are enabled", func() {
+	It("prevents removing the replication slot section when highAvailability is enabled", func() {
 		oldCluster := &apiv1.Cluster{
 			Spec: apiv1.ClusterSpec{
 				ImageName: versions.DefaultImageName,
@@ -3590,6 +3596,30 @@ var _ = Describe("validation of replication slots configuration", func() {
 						Enabled:    ptr.To(true),
 						SlotPrefix: "_test_",
 					},
+					SynchronizeReplicas: &apiv1.SynchronizeReplicasConfiguration{
+						Enabled: ptr.To(false),
+					},
+				},
+			},
+		}
+		oldCluster.Default()
+
+		newCluster := oldCluster.DeepCopy()
+		newCluster.Spec.ReplicationSlots = nil
+		Expect(v.validateReplicationSlotsChange(newCluster, oldCluster)).To(HaveLen(1))
+	})
+
+	It("prevents removing the replication slot section when synchronizeReplicas is enabled", func() {
+		oldCluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				ImageName: versions.DefaultImageName,
+				ReplicationSlots: &apiv1.ReplicationSlotsConfiguration{
+					HighAvailability: &apiv1.ReplicationSlotsHAConfiguration{
+						Enabled: ptr.To(false),
+					},
+					SynchronizeReplicas: &apiv1.SynchronizeReplicasConfiguration{
+						Enabled: ptr.To(true),
+					},
 				},
 			},
 		}
@@ -3600,6 +3630,28 @@ var _ = Describe("validation of replication slots configuration", func() {
 		Expect(v.validateReplicationSlotsChange(newCluster, oldCluster)).To(HaveLen(1))
 	})
 
+	It("prevents removing the replication slot section when both features are enabled", func() {
+		oldCluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				ImageName: versions.DefaultImageName,
+				ReplicationSlots: &apiv1.ReplicationSlotsConfiguration{
+					HighAvailability: &apiv1.ReplicationSlotsHAConfiguration{
+						Enabled:    ptr.To(true),
+						SlotPrefix: "_test_",
+					},
+					SynchronizeReplicas: &apiv1.SynchronizeReplicasConfiguration{
+						Enabled: ptr.To(true),
+					},
+				},
+			},
+		}
+		oldCluster.Default()
+
+		newCluster := oldCluster.DeepCopy()
+		newCluster.Spec.ReplicationSlots = nil
+		Expect(v.validateReplicationSlotsChange(newCluster, oldCluster)).To(HaveLen(2))
+	})
+
 	It("allows disabling the replication slots", func() {
 		oldCluster := &apiv1.Cluster{
 			Spec: apiv1.ClusterSpec{
@@ -3663,6 +3715,72 @@ var _ = Describe("validation of replication slots configuration", func() {
 		Expect(errors).To(BeEmpty())
 	})
 
+	It("prevents removing the synchronizeReplicas section when synchronizeReplicas is enabled", func() {
+		oldCluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				ImageName: versions.DefaultImageName,
+				ReplicationSlots: &apiv1.ReplicationSlotsConfiguration{
+					HighAvailability: &apiv1.ReplicationSlotsHAConfiguration{
+						Enabled: ptr.To(false),
+					},
+					SynchronizeReplicas: &apiv1.SynchronizeReplicasConfiguration{
+						Enabled: ptr.To(true),
+					},
+				},
+			},
+		}
+		oldCluster.Default()
+
+		newCluster := oldCluster.DeepCopy()
+		newCluster.Spec.ReplicationSlots.SynchronizeReplicas = nil
+		errs := v.validateReplicationSlotsChange(newCluster, oldCluster)
+		Expect(errs).To(HaveLen(1))
+		Expect(errs[0].Error()).To(ContainSubstring("Cannot remove"))
+		Expect(errs[0].Error()).To(ContainSubstring("synchronizeReplicas"))
+	})
+
+	It("allows disabling the synchronizeReplicas feature", func() {
+		oldCluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				ImageName: versions.DefaultImageName,
+				ReplicationSlots: &apiv1.ReplicationSlotsConfiguration{
+					HighAvailability: &apiv1.ReplicationSlotsHAConfiguration{
+						Enabled: ptr.To(false),
+					},
+					SynchronizeReplicas: &apiv1.SynchronizeReplicasConfiguration{
+						Enabled: ptr.To(true),
+					},
+				},
+			},
+		}
+		oldCluster.Default()
+
+		newCluster := oldCluster.DeepCopy()
+		newCluster.Spec.ReplicationSlots.SynchronizeReplicas.Enabled = ptr.To(false)
+		Expect(v.validateReplicationSlotsChange(newCluster, oldCluster)).To(BeEmpty())
+	})
+
+	It("allows removing synchronizeReplicas section when synchronizeReplicas is disabled", func() {
+		oldCluster := &apiv1.Cluster{
+			Spec: apiv1.ClusterSpec{
+				ImageName: versions.DefaultImageName,
+				ReplicationSlots: &apiv1.ReplicationSlotsConfiguration{
+					HighAvailability: &apiv1.ReplicationSlotsHAConfiguration{
+						Enabled: ptr.To(false),
+					},
+					SynchronizeReplicas: &apiv1.SynchronizeReplicasConfiguration{
+						Enabled: ptr.To(false),
+					},
+				},
+			},
+		}
+		oldCluster.Default()
+
+		newCluster := oldCluster.DeepCopy()
+		newCluster.Spec.ReplicationSlots.SynchronizeReplicas = nil
+		Expect(v.validateReplicationSlotsChange(newCluster, oldCluster)).To(BeEmpty())
+	})
+
 	It("returns no errors when synchronizeLogicalDecoding is disabled", func() {
 		cluster := &apiv1.Cluster{
 			Spec: apiv1.ClusterSpec{