[DPE-10370] fix(backups): reject standby backup actions and harden async replication (#1587)

* fix(backups): reject standby backup actions and harden async replication

Fail create-backup, list-backups, and restore on standby clusters with
explicit guidance to run those actions on the primary cluster.

Also guard async replication relation handling when no remote units are
present yet, preventing uncaught StopIteration during create-replication.

Add unit coverage for both fixes, add a Juju3 Ceph-backed integration test
for async replication backup behavior, and switch microceph TLS setup to a
non-loopback host IP.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test(integration): simplify action assertions for Juju 3

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* fix(async-replication): correct re-emission of relation-changed event

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test(unit): fix async replication mocks

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test(integration): migrate async replication backup test to jubilant

Re-implement the standby-backup integration test on jubilant under the
high_availability/ package, mirroring the jubilant implementation from

- Replace ha_tests/test_async_replication_backups_ceph.py with a
  jubilant version under high_availability/, plus the package __init__
- Add high_availability_helpers_new.py with leader/units/primary
  lookups and wait_for_apps_status
- Add a jubilant `juju` fixture to the integration conftest and depend
  on jubilant ^1.8.0
- Point the spread task at the new module path and drop the -juju29
  variant
- Use REPLICATION_CONSUMER_RELATION/REPLICATION_OFFER_RELATION constants
  in _is_standby_cluster instead of "replication"/"replication-offer"
- Reword async replication action failures from "pod" to "unit"
  addresses and update the matching unit test

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test(integration): fix async replication ceph backup test

Get test_standby_backup_rejected_with_clear_message running green
against real Juju + Ceph:

- Pass the s3-integrator config at deploy time and wait for the app to
  reach blocked status (10m) before syncing credentials, instead of
  configuring it separately while the unit is still provisioning
- Run create-backup on the primary unit: without TLS the charm only
  permits backups on the primary, and this test does not enable TLS
  (drop the now-unused get_app_units import)
- Give the second_model fixture its own Juju() instance so it stops
  reassigning the shared temp-model's model name, which broke
  jubilant.temp_model teardown and leaked the model

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test(integration): exclude async replication backup test from Juju 2.9

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

---------

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

Author: marceloneppel

poetry.lock

@@ -66,6 +66,7 @@ pytest = "^9.1.0"
 pytest-operator = "^0.43.2"
 # renovate caret doesn't work: https://github.com/renovatebot/renovate/issues/26940
 juju = "<=3.6.1.3"
+jubilant = "^1.8.0"
 boto3 = "*"
 tenacity = "*"
 landscape-api-py3 = "^0.9.0"

pyproject.toml

@@ -45,6 +45,7 @@
     PGBACKREST_LOGS_PATH,
     POSTGRESQL_DATA_PATH,
 )
+from relations.async_replication import REPLICATION_CONSUMER_RELATION, REPLICATION_OFFER_RELATION
 
 logger = logging.getLogger(__name__)
 
@@ -60,6 +61,18 @@
     FAILED_TO_ACCESS_CREATE_BUCKET_ERROR_MESSAGE,
     FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE,
 ]
+STANDBY_CLUSTER_CREATE_BACKUP_ERROR_MESSAGE = (
+    "Backups are not supported on a standby cluster. "
+    "Run create-backup on the primary cluster instead."
+)
+STANDBY_CLUSTER_LIST_BACKUPS_ERROR_MESSAGE = (
+    "Backups are not supported on a standby cluster. "
+    "Run list-backups on the primary cluster instead."
+)
+STANDBY_CLUSTER_RESTORE_ERROR_MESSAGE = (
+    "Restoring backups is not supported on a standby cluster. "
+    "Run restore on the primary cluster instead."
+)
 
 
 class ListBackupsError(Exception):
@@ -153,6 +166,9 @@ def _can_initialise_stanza(self) -> bool:
 
     def _can_unit_perform_backup(self) -> tuple[bool, str | None]:
         """Validates whether this unit can perform a backup."""
+        if self._is_standby_cluster():
+            return False, STANDBY_CLUSTER_CREATE_BACKUP_ERROR_MESSAGE
+
         if self.charm.is_blocked:
             return False, "Unit is in a blocking state"
 
@@ -183,6 +199,16 @@ def _can_unit_perform_backup(self) -> tuple[bool, str | None]:
 
         return self._are_backup_settings_ok()
 
+    def _is_standby_cluster(self) -> bool:
+        """Return whether this unit belongs to a standby cluster."""
+        if (
+            self.model.get_relation(REPLICATION_CONSUMER_RELATION) is None
+            and self.model.get_relation(REPLICATION_OFFER_RELATION) is None
+        ):
+            return False
+
+        return not self.charm.async_replication.is_primary_cluster()
+
     def can_use_s3_repository(self) -> tuple[bool, str | None]:
         """Returns whether the charm was configured to use another cluster repository."""
         # Check model uuid
@@ -1060,6 +1086,11 @@ def _run_backup(
 
     def _on_list_backups_action(self, event) -> None:
         """List the previously created backups."""
+        if self._is_standby_cluster():
+            logger.warning(STANDBY_CLUSTER_LIST_BACKUPS_ERROR_MESSAGE)
+            event.fail(STANDBY_CLUSTER_LIST_BACKUPS_ERROR_MESSAGE)
+            return
+
         are_backup_settings_ok, validation_message = self._are_backup_settings_ok()
         if not are_backup_settings_ok:
             logger.warning(validation_message)
@@ -1239,6 +1270,11 @@ def _pre_restore_checks(self, event: ActionEvent) -> bool:
         Returns:
             a boolean indicating whether restore should be run.
         """
+        if self._is_standby_cluster():
+            logger.error(f"Restore failed: {STANDBY_CLUSTER_RESTORE_ERROR_MESSAGE}")
+            event.fail(STANDBY_CLUSTER_RESTORE_ERROR_MESSAGE)
+            return False
+
         are_backup_settings_ok, validation_message = self._are_backup_settings_ok()
         if not are_backup_settings_ok:
             logger.error(f"Restore failed: {validation_message}")

src/backups.py

@@ -434,14 +434,25 @@ def _handle_replication_change(self, event: ActionEvent) -> bool:
             return False
 
         relation = self._relation
+        if relation is None:
+            event.fail("Replication relation not found")
+            return False
 
-        # Check if all units from the other cluster  published their pod IPs in the relation data.
-        # If not, fail the action telling that all units must publish their pod addresses in the
+        # Ensure the relation has at least one remote unit before trying to process unit data.
+        remote_units = [unit for unit in relation.units if unit.app == relation.app]
+        if len(remote_units) == 0:
+            event.fail(
+                "All units from the other cluster must publish their unit addresses in the relation data."
+            )
+            return False
+
+        # Check if all units from the other cluster published their IPs in the relation data.
+        # If not, fail the action telling that all units must publish their unit addresses in the
         # relation data.
-        for unit in relation.units:
+        for unit in remote_units:
             if "unit-address" not in relation.data[unit]:
                 event.fail(
-                    "All units from the other cluster must publish their pod addresses in the relation data."
+                    "All units from the other cluster must publish their unit addresses in the relation data."
                 )
                 return False
 
@@ -648,12 +659,20 @@ def _primary_cluster_endpoint(self) -> str:
 
     def _re_emit_async_relation_changed_event(self) -> None:
         """Re-emit the async relation changed event."""
-        relation = self._relation
-        getattr(self.charm.on, f"{relation.name.replace('-', '_')}_relation_changed").emit(
-            relation,
-            app=relation.app,
-            unit=next(unit for unit in relation.units if unit.app == relation.app),
-        )
+        if relation := self._relation:
+            relation_unit = next(
+                (unit for unit in relation.units if unit.app == relation.app), None
+            )
+            if relation_unit is None:
+                logger.debug(
+                    "Skipping re-emitting relation-changed event: no related units found yet."
+                )
+                return
+            getattr(self.charm.on, f"{relation.name.replace('-', '_')}_relation_changed").emit(
+                relation,
+                app=relation.app,
+                unit=relation_unit,
+            )
 
     def _reinitialise_pgdata(self) -> None:
         """Reinitialise the pgdata folder."""

src/relations/async_replication.py

@@ -4,11 +4,11 @@
 import json
 import logging
 import os
-import socket
 import subprocess
 import uuid
 
 import boto3
+import jubilant
 import pytest
 from pytest_operator.plugin import OpsTest
 
@@ -21,6 +21,17 @@
 logger = logging.getLogger(__name__)
 
 
+def _get_non_loopback_host_ip() -> str:
+    """Return the host IP reachable by LXD containers."""
+    output = subprocess.run(
+        ["ip", "-4", "route", "get", "1.1.1.1"],
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout
+    return output.split("src", maxsplit=1)[1].split()[0]
+
+
 @pytest.fixture(scope="session")
 def charm():
     # Return str instead of pathlib.Path since python-libjuju's model.deploy(), juju deploy, and
@@ -101,6 +112,20 @@ async def gcp_cloud_configs(ops_test: OpsTest) -> None:
     cleanup_cloud(config, credentials)
 
 
+@pytest.fixture(scope="module")
+def juju(request: pytest.FixtureRequest):
+    """Pytest fixture that wraps jubilant.temp_model."""
+    model = request.config.getoption("--model")
+    keep_models = bool(request.config.getoption("--keep-models"))
+
+    if model:
+        juju = jubilant.Juju(model=model)
+        yield juju
+    else:
+        with jubilant.temp_model(keep=keep_models) as juju:
+            yield juju
+
+
 @dataclasses.dataclass(frozen=True)
 class ConnectionInformation:
     access_key_id: str
@@ -151,7 +176,7 @@ def microceph():
         ],
         check=True,
     )
-    host_ip = socket.gethostbyname(socket.gethostname())
+    host_ip = _get_non_loopback_host_ip()
     subprocess.run(
         f'echo "subjectAltName = IP:{host_ip}" > ./extfile.cnf',
         shell=True,
@@ -209,7 +234,7 @@ def microceph():
     key_id = key["access_key"]
     secret_key = key["secret_key"]
     logger.info("Set up microceph")
-    host_ip = socket.gethostbyname(socket.gethostname())
+    host_ip = _get_non_loopback_host_ip()
     result = subprocess.run(
         "base64 -w0 ./ca.crt", shell=True, check=True, stdout=subprocess.PIPE, text=True
     )

tests/integration/conftest.py

@@ -0,0 +1,2 @@
+# Copyright 2026 Canonical Ltd.
+# See LICENSE file for licensing details.

tests/integration/high_availability/__init__.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+# Copyright 2026 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+from collections.abc import Callable
+
+import jubilant
+from jubilant import Juju
+from jubilant.statustypes import Status, UnitStatus
+
+MINUTE_SECS = 60
+
+JujuModelStatusFn = Callable[[Status], bool]
+JujuAppsStatusFn = Callable[[Status, str], bool]
+
+
+def get_app_leader(juju: Juju, app_name: str) -> str:
+    """Get the leader unit for the given application."""
+    app_status = juju.status().apps[app_name]
+    for name, status in app_status.units.items():
+        if status.leader:
+            return name
+
+    raise Exception("No leader unit found")
+
+
+def get_app_units(juju: Juju, app_name: str) -> dict[str, UnitStatus]:
+    """Get the units for the given application."""
+    return juju.status().apps[app_name].units
+
+
+def get_db_primary_unit(juju: Juju, app_name: str) -> str:
+    """Get the current primary node of the cluster."""
+    postgresql_primary = get_app_leader(juju, app_name)
+    task = juju.run(unit=postgresql_primary, action="get-primary", wait=5 * MINUTE_SECS)
+    task.raise_on_failure()
+
+    primary = task.results.get("primary")
+    if primary != "None":
+        return primary
+
+    raise Exception("No primary node found")
+
+
+def wait_for_apps_status(jubilant_status_func: JujuAppsStatusFn, *apps: str) -> JujuModelStatusFn:
+    """Wait for Juju agents to be idle and apps to reach a target status."""
+    return lambda status: all((
+        jubilant.all_agents_idle(status, *apps),
+        jubilant_status_func(status, *apps),
+    ))