fix(topology): relaunch the observer only when its inputs change

src/literals.py

@@ -28,6 +28,7 @@
 TOPOLOGY_OBSERVER_LOG_FILENAME = "topology_observer.log"
 TOPOLOGY_OBSERVER_TLS_CA_FILENAME = "valkey_ca.pem"
 TOPOLOGY_OBSERVER_PID_FILENAME = "topology_observer.pid"
+TOPOLOGY_OBSERVER_SIGNATURE_FILENAME = "topology_observer.signature"
 
 PEER_RELATION = "valkey-peers"
 STATUS_PEERS_RELATION = "status-peers"

src/managers/topology.py

@@ -8,6 +8,7 @@
 import os
 import signal
 import subprocess
+from hashlib import sha256
 from pathlib import Path
 from sys import version_info
 
@@ -18,6 +19,7 @@
     SENTINEL_TLS_PORT,
     TOPOLOGY_OBSERVER_LOG_FILENAME,
     TOPOLOGY_OBSERVER_PID_FILENAME,
+    TOPOLOGY_OBSERVER_SIGNATURE_FILENAME,
     TOPOLOGY_OBSERVER_TLS_CA_FILENAME,
     CharmUsers,
 )
@@ -50,16 +52,64 @@ def _pid_file_path(self) -> Path:
         """Return the path to the topology observer pid file."""
         return self.state.charm.charm_dir / TOPOLOGY_OBSERVER_PID_FILENAME
 
+    @property
+    def _signature_file_path(self) -> Path:
+        """Return the path to the topology observer signature file."""
+        return self.state.charm.charm_dir / TOPOLOGY_OBSERVER_SIGNATURE_FILENAME
+
+    @property
+    def _observer_hosts(self) -> str:
+        """Return the Sentinel host list the observer subprocess is launched with."""
+        started_servers = [
+            unit.get_endpoint(self.state.substrate)
+            for unit in self.state.servers
+            if unit.is_active
+        ]
+        port = SENTINEL_TLS_PORT if self.state.unit_server.is_tls_enabled else SENTINEL_PORT
+        return ",".join(sorted([f"{server}:{port}" for server in started_servers]))
+
+    def observer_signature(self) -> str:
+        """Return a digest of the arguments the observer subprocess is launched with.
+
+        Recorded next to the PID so a re-delivered event can tell an observer that
+        is already watching the right topology from one that has to be relaunched.
+        Hashed because the Sentinel password is one of those arguments.
+        """
+        parts = [
+            self._observer_hosts,
+            str(self.state.unit_server.is_tls_enabled),
+            CharmUsers.SENTINEL_CHARM_ADMIN.value,
+            self.state.cluster.internal_users_credentials.get(
+                CharmUsers.SENTINEL_CHARM_ADMIN.value, ""
+            ),
+        ]
+        if self.state.unit_server.is_tls_enabled:
+            # the observer is handed its own copy of the CA, so a rotation has to relaunch
+            # it; an unreadable CA yields "" and reads as a change, which is the safe way
+            # round -- start_observer surfaces the real error when it re-reads the file
+            try:
+                parts.append(self.workload.read_file(self.workload.tls_paths.client_ca))
+            except Exception:  # noqa: BLE001
+                logger.debug("Could not read the client CA while signing the observer")
+                parts.append("")
+
+        return sha256("|".join(parts).encode()).hexdigest()
+
+    def _is_observer_running(self) -> bool:
+        """Return whether the recorded observer process is still alive."""
+        if (observer_pid := self._read_observer_pid()) == 0:
+            return False
+        try:
+            os.kill(observer_pid, 0)
+            return True
+        except OSError:
+            logger.debug("Topology observer not running")
+            return False
+
     def start_observer(self) -> None:
         """Start the topology observer as a subprocess."""
-        if (observer_pid := self._read_observer_pid()) != 0:
-            try:
-                # check if the process already runs
-                os.kill(int(observer_pid), 0)
-                return
-            except OSError:
-                logger.debug("Topology observer not running")
-                pass
+        if self._is_observer_running():
+            return
 
         # Generate the venv path based on the existing lib path
         env = os.environ.copy()
@@ -79,13 +129,9 @@ def start_observer(self) -> None:
                 break
 
         # Gather Valkey hosts for connection
-        started_servers = [
-            unit.get_endpoint(self.state.substrate)
-            for unit in self.state.servers
-            if unit.is_active
-        ]
-        port = SENTINEL_TLS_PORT if self.state.unit_server.is_tls_enabled else SENTINEL_PORT
-        hosts = ",".join(sorted([f"{server}:{port}" for server in started_servers]))
+        hosts = self._observer_hosts
+        # captured before the spawn so the record matches what the process was given
+        signature = self.observer_signature()
 
         if self.state.unit_server.is_tls_enabled:
             # Store current TLS CA cert on operator container
@@ -113,10 +159,13 @@ def start_observer(self) -> None:
         ).pid
 
         self._pid_file_path.write_text(str(pid))
+        self._signature_file_path.write_text(signature)
         logging.info(f"Started topology observer process with PID {pid}")
 
     def stop_observer(self) -> None:
         """Stop the topology observer."""
+        self._signature_file_path.unlink(missing_ok=True)
+
         if (observer_pid := self._read_observer_pid()) == 0:
             logger.debug("Topology observer already stopped")
             return
@@ -128,16 +177,37 @@ def stop_observer(self) -> None:
         except OSError:
             pass
         finally:
-            self._pid_file_path.unlink()
+            self._pid_file_path.unlink(missing_ok=True)
 
     def restart_observer(self) -> None:
-        """Stop and start the topology observer to pickup host changes."""
+        """Relaunch the topology observer if it is gone or watching a stale topology.
+
+        The leader calls this on every peer relation-changed, and the observer only
+        reports a primary change against the one it saw last -- a value it keeps in
+        memory. Relaunching a healthy observer therefore throws that value away and
+        leaves the cluster unwatched for the length of a Python start-up, so a
+        failover landing in that window is never dispatched. Leave it alone unless
+        an argument it was launched with has actually changed.
+        """
+        if self._is_observer_running() and self._read_observer_signature() == (
+            self.observer_signature()
+        ):
+            logger.debug("Topology observer already watching the current topology")
+            return
+
         self.stop_observer()
         self.start_observer()
 
     def _read_observer_pid(self) -> int:
         """Read the pid file of the topology observer and return the pid, or 0 if none."""
         try:
             return int(self._pid_file_path.read_text())
-        except (FileNotFoundError, PermissionError):
+        except (FileNotFoundError, PermissionError, ValueError):
             return 0
+
+    def _read_observer_signature(self) -> str:
+        """Read the signature of the running observer, or "" if there is none recorded."""
+        try:
+            return self._signature_file_path.read_text()
+        except (FileNotFoundError, PermissionError):
+            return ""

tests/integration/clients/test_client_relation.py

@@ -28,6 +28,8 @@
     get_cluster_addresses,
     get_password,
     get_primary_ip,
+    has_leader_settled_since,
+    utc_now,
     wait_for_failover,
 )
 
@@ -186,6 +188,10 @@ def test_failover_topology_update(juju: jubilant.Juju) -> None:
     old_primary_ip = get_primary_ip(juju, APP_NAME)
     logger.info("Initiate failover through Sentinel %s", ip_address)
 
+    # Anchor before the failover: the gate below has to wait for the charm to react to it,
+    # and until the leader is woken it still reports the idle it entered beforehand.
+    failover_requested_at = utc_now()
+
     failover_result = exec_valkey_cli(
         hostname=ip_address,
         username=CharmUsers.SENTINEL_CHARM_ADMIN,
@@ -197,8 +203,12 @@ def test_failover_topology_update(juju: jubilant.Juju) -> None:
     assert failover_result == "OK", "Failover not successful"
 
     wait_for_failover(juju, APP_NAME, old_primary_ip=old_primary_ip, unit_count=NUM_UNITS)
+    # Sentinel has switched, but the client still reaches Valkey through whatever the charm
+    # last published; `topology_changed` is what re-points it, and only the leader handles it.
     juju.wait(
-        lambda status: are_agents_idle(status, APP_NAME, idle_period=60, unit_count=NUM_UNITS),
+        lambda status: has_leader_settled_since(
+            status, APP_NAME, since=failover_requested_at, idle_period=60
+        ),
         timeout=600,
     )
 

tests/integration/helpers.py

@@ -7,7 +7,7 @@
 import re
 import subprocess
 from contextlib import contextmanager
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import List, Literal, NamedTuple
 
@@ -193,15 +193,48 @@ def are_agents_idle(
     )
 
 
+def utc_now() -> datetime:
+    """Return the current UTC time, naive, to compare against Juju status timestamps.
+
+    Juju reports timestamps in UTC and they are parsed with `ignoretz`, so the value they
+    are compared against has to be UTC too. `datetime.now()` is local: anywhere east of
+    Greenwich it runs ahead, and every elapsed-time check passes by that offset alone.
+    """
+    return datetime.now(timezone.utc).replace(tzinfo=None)
+
+
 def _check_apps_idle_period(status: jubilant.Status, *apps: str, idle_period: int) -> bool:
     return all(
-        parse(unit.juju_status.since, ignoretz=True) + timedelta(seconds=idle_period)
-        < datetime.now()
+        parse(unit.juju_status.since, ignoretz=True) + timedelta(seconds=idle_period) < utc_now()
         for app in apps
         for unit in status.get_units(app).values()
     )
 
 
+def has_leader_settled_since(
+    status: jubilant.Status, app: str, since: datetime, idle_period: int = 0
+) -> bool:
+    """Whether the leader has run a hook and gone idle again after `since` (naive UTC).
+
+    An `idle_period` on its own cannot express "the charm has reacted to X". An agent that
+    has not been woken yet still carries the idle timestamp it had before X, and that
+    timestamp already satisfies any period, so the wait returns on its first poll and the
+    test races ahead of the charm. Anchoring on `since` waits for the leader to actually
+    pick the event up; the period then lets any deferral it left behind drain.
+
+    Only the leader is considered: handlers such as `topology_changed` are leader-only and
+    write no peer data, so the other units are never woken and would never settle.
+    """
+    for unit in status.get_units(app).values():
+        if not unit.leader:
+            continue
+        if unit.juju_status.current != "idle":
+            return False
+        idle_since = parse(unit.juju_status.since, ignoretz=True)
+        return idle_since >= since and idle_since + timedelta(seconds=idle_period) < utc_now()
+    return False
+
+
 def verify_unit_count(
     status: jubilant.Status, *apps: str, unit_count: int | dict[str, int] | None = None
 ):