[DPE-10397] The watcher now handles 4, 6, … units (2 units worked previously)

src/raft_controller.py

@@ -4,8 +4,8 @@
 """Raft controller management for PostgreSQL watcher.
 
 This module manages a Patroni raft_controller node that participates in
-consensus without running PostgreSQL, providing the necessary third vote
-for quorum in 2-node PostgreSQL clusters.
+consensus without running PostgreSQL, providing the extra vote that keeps
+quorum odd in even-sized (2, 4, 6, ...) PostgreSQL clusters.
 
 Uses Patroni's own ``patroni_raft_controller`` from the charmed-postgresql
 snap, which is the same battle-tested Raft implementation used by the

src/relations/watcher_requirer.py

@@ -6,7 +6,8 @@
 This module handles the watcher (requirer) side of the relation, used when the
 charm is deployed with role=watcher. It connects to one or more PostgreSQL
 applications (which provide the watcher-offer relation) and participates in
-Raft consensus as a lightweight witness for stereo mode (2-node clusters).
+Raft consensus as a lightweight witness whenever the PostgreSQL unit count is
+even (2, 4, 6, ...), standing down at odd counts of three or more.
 
 Multi-cluster support:
 - Each watcher relation gets its own RaftController instance
@@ -354,6 +355,51 @@ def _update_unit_address_if_changed(self) -> None:
                 raft_controller.cleanup_raft_cluster(watcher_addr, raft_password, partner_addrs)
         self.charm.app_peer_data["unit-address"] = new_address
 
+    def _reconcile_relation_raft(self, relation: Relation) -> tuple[bool, bool, str | None]:
+        """Reconcile the Raft service state for a single relation.
+
+        Stands the watcher down (stops the service and leaves the quorum) when
+        the provider disabled it or the PostgreSQL unit count is odd >= 3, and
+        self-heals the service and Raft membership when the watcher must vote.
+
+        Returns:
+            Tuple of (connected, stood_down, warning) where warning carries the
+            odd-unit-count message when the watcher stands down because of it.
+        """
+        port = self._get_port_for_relation(relation.id)
+        password = self._get_raft_password(relation)
+        raft_controller = RaftController(self.charm, instance_id=f"rel{relation.id}")
+        raft_status = raft_controller.get_status(port, password)
+        connected = bool(raft_status.get("connected"))
+
+        partner_addrs = self._get_raft_partner_addrs(relation)
+        provider_disabled = self._is_disabled(relation)
+        stand_down = provider_disabled or not self._should_watcher_vote(partner_addrs)
+
+        if password and stand_down:
+            # Stop the service before removing the raft member, otherwise the
+            # still-running pysyncobj node can rejoin before the service stops
+            if relation.data[self.charm.app].get("raft-status") != "disabled":
+                raft_controller.remove_service()
+                raft_controller.remove_raft_member(
+                    f"{self.unit_ip}:{port}", password, partner_addrs
+                )
+                relation.data[self.charm.app]["raft-status"] = "disabled"
+            warning = None if provider_disabled else self._odd_units_warning(relation)
+            return False, True, warning
+
+        if password and partner_addrs:
+            # Self-heal: rejoin the quorum if the relation-changed event that
+            # should have re-enabled the watcher was missed
+            if not raft_status.get("running"):
+                raft_controller.configure(
+                    port, self.unit_ip, partner_addrs, password, self._get_patroni_cas(relation)
+                )
+            if service_running(raft_controller.service_name):
+                relation.data[self.charm.app]["raft-status"] = "connected"
+                connected = True
+        return connected, provider_disabled, None
+
     def _on_update_status(self, event: UpdateStatusEvent) -> None:
         """Handle update status event in watcher mode."""
         if not self.charm.unit.is_leader():
@@ -375,29 +421,12 @@ def _on_update_status(self, event: UpdateStatusEvent) -> None:
         info_warnings: list[str] = []
 
         for relation in relations:
-            port = self._get_port_for_relation(relation.id)
-            password = self._get_raft_password(relation)
-            raft_controller = RaftController(self.charm, instance_id=f"rel{relation.id}")
-            raft_status = raft_controller.get_status(port, password)
-            disabled = disabled or self._is_disabled(relation)
-            connected_count += 1 if raft_status.get("connected") else 0
-
-            pg_endpoints = self._get_raft_partner_addrs(relation)
-            total_endpoints += len(pg_endpoints)
-            partner_addrs = self._get_raft_partner_addrs(relation)
-
-            if password and not self._should_watcher_vote(partner_addrs):
-                cluster_name = self._get_cluster_name(relation)
-                raft_controller.remove_raft_member(
-                    f"{self.unit_ip}:{port}", password, pg_endpoints
-                )
-                info_warnings.append(
-                    f"WARNING: cluster '{cluster_name}' has odd number units;"
-                    " adding a watcher creates even Raft membership,"
-                    " which degrades partition tolerance"
-                )
-                raft_controller.remove_service()
-                disabled = True
+            connected, stood_down, warning = self._reconcile_relation_raft(relation)
+            total_endpoints += len(self._get_raft_partner_addrs(relation))
+            connected_count += 1 if connected else 0
+            disabled = disabled or stood_down
+            if warning:
+                info_warnings.append(warning)
 
             az_warning = self._check_az_colocation(relation)
             if az_warning:
@@ -407,25 +436,37 @@ def _on_update_status(self, event: UpdateStatusEvent) -> None:
             self.charm.unit.status = WaitingStatus("Connecting to Raft cluster")
             return
 
-        cluster_count = len(relations)
-        msg = (
-            f"Raft connected, monitoring {total_endpoints} PostgreSQL endpoints"
-            if cluster_count == 1
-            else (
+        self.charm.unit.status = self._summary_status(
+            len(relations), connected_count, total_endpoints, az_warnings, info_warnings
+        )
+
+    def _summary_status(
+        self,
+        cluster_count: int,
+        connected_count: int,
+        total_endpoints: int,
+        az_warnings: list[str],
+        info_warnings: list[str],
+    ) -> ActiveStatus | BlockedStatus:
+        """Compose the unit status summarizing all watcher relations."""
+        if connected_count == 0:
+            msg = f"Watcher standing down, monitoring {total_endpoints} PostgreSQL endpoints"
+        elif cluster_count == 1:
+            msg = f"Raft connected, monitoring {total_endpoints} PostgreSQL endpoints"
+        else:
+            msg = (
                 f"Raft connected to {connected_count}/{cluster_count} clusters, "
                 f"monitoring {total_endpoints} PostgreSQL endpoints"
             )
-        )
 
         # AZ co-location blocks in production; odd-count warnings never block
         if az_warnings and self.charm.config.profile == "production":
-            self.charm.unit.status = BlockedStatus("AZ co-location: " + "; ".join(az_warnings))
-            return
+            return BlockedStatus("AZ co-location: " + "; ".join(az_warnings))
 
         if all_warnings := az_warnings + info_warnings:
             msg += "; " + "; ".join(all_warnings)
 
-        self.charm.unit.status = ActiveStatus(msg)
+        return ActiveStatus(msg)
 
     def _check_az_colocation(self, relation: Relation) -> str | None:
         """Check if the watcher is in the same AZ as any PostgreSQL unit.
@@ -470,6 +511,14 @@ def _should_watcher_vote(self, partner_addrs: list[str]) -> bool:
         pg_num = len(partner_addrs)
         return pg_num < 3 or pg_num % 2 == 0
 
+    def _odd_units_warning(self, relation: Relation) -> str:
+        """Warning shown while the watcher stands down for an odd-sized cluster."""
+        return (
+            f"WARNING: cluster '{self._get_cluster_name(relation)}' has odd number units;"
+            " adding a watcher creates even Raft membership,"
+            " which degrades partition tolerance"
+        )
+
     def _on_watcher_relation_changed(
         self, event: RelationChangedEvent | SecretChangedEvent
     ) -> None:
@@ -494,7 +543,7 @@ def _on_watcher_relation_changed(
                 partner_addrs := self._get_raft_partner_addrs(relation)
             ):
                 logger.debug("Raft details are not yet available")
-                return
+                continue
 
             # Get or assign a port for this relation
             port = self._get_port_for_relation(relation.id)
@@ -509,7 +558,9 @@ def _on_watcher_relation_changed(
                     f"{self.unit_ip}:{port}", raft_password, partner_addrs
                 )
                 relation.data[self.charm.app]["raft-status"] = "disabled"
-                return
+                if not self._is_disabled(relation):
+                    self.charm.unit.status = ActiveStatus(self._odd_units_warning(relation))
+                continue
 
             raft_controller.configure(
                 port, unit_ip, partner_addrs, raft_password, self._get_patroni_cas(relation)

tests/integration/ha_tests/helpers.py

@@ -9,6 +9,7 @@
 import requests
 import yaml
 from juju.model import Model
+from pysyncobj.utility import TcpUtility
 from pytest_operator.plugin import OpsTest
 from tenacity import (
     RetryError,
@@ -19,6 +20,8 @@
     wait_fixed,
 )
 
+from constants import RAFT_PARTNER_PREFIX
+
 from ..helpers import (
     APPLICATION_NAME,
     get_password,
@@ -419,3 +422,107 @@ async def wait_network_restore(ops_test: OpsTest, unit_name: str, old_ip: str) -
     # Juju status too quickly.
     if (await get_ip_from_inside_the_unit(ops_test, unit_name)) == old_ip:
         raise Exception
+
+
+async def start_writes(ops_test: OpsTest) -> None:
+    """Start continuous writes to PostgreSQL (assumes relation already exists)."""
+    for attempt in Retrying(stop=stop_after_delay(60 * 5), wait=wait_fixed(3), reraise=True):
+        with attempt:
+            action = (
+                await ops_test.model
+                .applications[APPLICATION_NAME]
+                .units[0]
+                .run_action("start-continuous-writes")
+            )
+            await action.wait()
+            assert action.results["result"] == "True", "Unable to create continuous_writes table"
+
+
+async def verify_raft_cluster_health(
+    ops_test: OpsTest,
+    db_app_name: str,
+    watcher_app_name: str,
+    expected_members: int = 3,
+    check_watcher_ip: bool = True,
+    expect_watcher_absent: bool = False,
+) -> None:
+    """Verify that the Raft cluster has the expected number of members and quorum.
+
+    This function checks that all PostgreSQL units see the expected number of
+    Raft members (including the watcher) and have quorum. This is critical
+    after watcher re-deployment to ensure the cluster is properly formed.
+
+    Args:
+        ops_test: The OpsTest instance.
+        db_app_name: The PostgreSQL application name.
+        watcher_app_name: The watcher application name.
+        expected_members: Expected number of Raft members (default 3 for stereo mode).
+        check_watcher_ip: Whether to verify the watcher IP in Raft status (default True).
+            Set to False after network isolation tests where watcher may have been
+            redeployed with a new IP that isn't yet in the Raft configuration.
+        expect_watcher_absent: When True, assert that the watcher IP is NOT a Raft
+            member (the watcher stood down, e.g. with an odd number of PostgreSQL
+            units). Takes precedence over check_watcher_ip.
+
+    Raises:
+        AssertionError: If the Raft cluster is not healthy.
+    """
+    logger.info(f"Verifying Raft cluster health with {expected_members} expected members")
+
+    # Get watcher address for verification using juju exec to avoid cached IPs
+    watcher_unit = ops_test.model.applications[watcher_app_name].units[0]
+    return_code, watcher_ip, _ = await ops_test.juju(
+        "exec", "--unit", watcher_unit.name, "--", "unit-get", "private-address"
+    )
+    assert return_code == 0, f"Failed to get watcher address from {watcher_unit.name}"
+    watcher_ip = watcher_ip.strip()
+
+    for attempt in Retrying(stop=stop_after_delay(180), wait=wait_fixed(5), reraise=True):
+        with attempt:
+            for unit in ops_test.model.applications[db_app_name].units:
+                # Get the Raft password from Patroni config using juju exec directly
+                # We need to avoid shell interpretation issues with run_command_on_unit
+                complete_command = [
+                    "exec",
+                    "--unit",
+                    unit.name,
+                    "--",
+                    "cat",
+                    "/var/snap/charmed-postgresql/current/etc/patroni/patroni.yaml",
+                ]
+                return_code, stdout, _ = await ops_test.juju(*complete_command)
+                assert return_code == 0, f"Failed to read patroni.yaml on {unit.name}"
+
+                conf = yaml.safe_load(stdout)
+                password = conf.get("raft", {}).get("password")
+                self_addr = conf.get("raft", {}).get("self_addr")
+                assert password, f"Could not find Raft password in patroni.yaml on {unit.name}"
+
+                # Check Raft status using the password
+                syncobj_util = TcpUtility(password=password, timeout=3)
+                status = syncobj_util.executeCommand(self_addr, ["status"])
+                logger.info(f"Raft status on {unit.name}: {status}")
+
+                # Verify quorum
+                assert status["has_quorum"] is True, f"Unit {unit.name} does not have Raft quorum"
+
+                assert status["partner_nodes_count"] + 1 == expected_members
+
+                member_ips = [
+                    key.split(":")[0].split(RAFT_PARTNER_PREFIX)[-1]
+                    for key in status
+                    if key.startswith(RAFT_PARTNER_PREFIX)
+                ]
+                if expect_watcher_absent:
+                    assert watcher_ip not in member_ips, (
+                        f"Watcher {watcher_ip} still in Raft cluster on {unit.name}"
+                    )
+                elif check_watcher_ip:
+                    # Verify watcher is in the cluster (if requested)
+                    # After network isolation tests, the watcher may have been redeployed
+                    # with a new IP that isn't yet updated in the Raft configuration
+                    assert watcher_ip in member_ips, (
+                        f"Watcher {watcher_ip} not found in Raft cluster on {unit.name}"
+                    )
+
+    logger.info("Raft cluster health verified successfully")