diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 510761f9a..4d1274ac1 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -68,7 +68,7 @@ jobs:
           - agent: 2.9.49  # renovate: juju-agent-pin-minor
             libjuju: ^2
             allure: false
-          - agent: 3.1.8  # renovate: juju-agent-pin-minor
+          - agent: 3.4.3  # renovate: juju-agent-pin-minor
             allure: true
     name: Integration test charm | ${{ matrix.juju.agent }}
     needs:
diff --git a/actions.yaml b/actions.yaml
index 5a7feebb7..35eb6c1ae 100644
--- a/actions.yaml
+++ b/actions.yaml
@@ -59,10 +59,20 @@ restore:
 pre-upgrade-check:
   description: Run necessary pre-upgrade checks and preparations before executing a charm refresh.
 
-promote-standby-cluster:
+create-replication:
   description: |
-    Promotes this cluster to become the leader in the cluster-set. Used for safe switchover or failover.
-    Must be run against the charm leader unit of a standby cluster.
+    Create replication between two related clusters.
+    This action is must be run on the offer side of the relation.
+  params:
+    name:
+      type: string
+      description: A (optional) name for this replication.
+      default: default
+
+promote-to-primary:
+  description: |
+    Promotes this cluster to become the primary in the cluster-set. Used for safe switchover or failover.
+    Can only be run against the charm leader unit of a standby cluster.
   params:
     cluster-set-name:
       type: string
@@ -83,24 +93,6 @@ recreate-cluster:
     each unit will be kept in blocked status. Recreating the cluster allows to rejoin the async replication
     relation, or usage as a standalone cluster.
 
-fence-writes:
-  description: |
-    Stops write traffic to a primary cluster of a ClusterSet.
-  params:
-    cluster-set-name:
-      type: string
-      description: |
-        The name of the cluster-set. Mandatory option, used for confirmation.
-
-unfence-writes:
-  description: |
-    Resumes write traffic to a primary cluster of a ClusterSet.
-  params:
-    cluster-set-name:
-      type: string
-      description: |
-        The name of the cluster-set. Mandatory option, used for confirmation.
-
 rejoin-cluster:
   description: |
     Rejoins an invalidated cluster to the cluster-set, after a previous failover or switchover.
diff --git a/lib/charms/mysql/v0/async_replication.py b/lib/charms/mysql/v0/async_replication.py
index ec89369fa..1e9fe1b23 100644
--- a/lib/charms/mysql/v0/async_replication.py
+++ b/lib/charms/mysql/v0/async_replication.py
@@ -1,13 +1,14 @@
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
 
-"""MySQL async replication module."""
+"""MySQL cluster-set async replication module."""
 
 import enum
 import logging
 import typing
 import uuid
 from functools import cached_property
+from time import sleep
 
 from charms.mysql.v0.mysql import (
     MySQLFencingWritesError,
@@ -20,8 +21,11 @@
     BlockedStatus,
     MaintenanceStatus,
     Relation,
+    RelationBrokenEvent,
+    RelationCreatedEvent,
     RelationDataContent,
     Secret,
+    SecretChangedEvent,
     SecretNotFoundError,
     WaitingStatus,
 )
@@ -51,11 +55,10 @@
 # The unique Charmhub library identifier, never change it
 LIBID = "4de21f1a022c4e2c87ac8e672ec16f6a"
 LIBAPI = 0
-LIBPATCH = 1
+LIBPATCH = 2
 
-PRIMARY_RELATION = "async-primary"
-REPLICA_RELATION = "async-replica"
-SECRET_LABEL = "async-secret"
+RELATION_OFFER = "replication-offer"
+RELATION_CONSUMER = "replication"
 
 
 class ClusterSetInstanceState(typing.NamedTuple):
@@ -86,10 +89,10 @@ def __init__(self, charm: "MySQLOperatorCharm", relation_name: str):
 
         # relation broken is observed on all units
         self.framework.observe(
-            self._charm.on[PRIMARY_RELATION].relation_broken, self.on_async_relation_broken
+            self._charm.on[RELATION_OFFER].relation_broken, self.on_async_relation_broken
         )
         self.framework.observe(
-            self._charm.on[REPLICA_RELATION].relation_broken, self.on_async_relation_broken
+            self._charm.on[RELATION_CONSUMER].relation_broken, self.on_async_relation_broken
         )
 
     @cached_property
@@ -106,10 +109,10 @@ def role(self) -> ClusterSetInstanceState:
 
         _, instance_role = self._charm._mysql.get_member_state()
 
-        if self.model.get_relation(REPLICA_RELATION):
-            relation_side = "replica"
+        if self.model.get_relation(RELATION_CONSUMER):
+            relation_side = RELATION_CONSUMER
         else:
-            relation_side = "primary"
+            relation_side = RELATION_OFFER
 
         return ClusterSetInstanceState(cluster_role, instance_role, relation_side)
 
@@ -123,13 +126,29 @@ def cluster_set_name(self) -> str:
         """Cluster set name."""
         return self._charm.app_peer_data["cluster-set-domain-name"]
 
-    def get_remote_relation_data(self, relation: Relation) -> Optional[RelationDataContent]:
-        """Remote data."""
-        if not relation.app:
+    @property
+    def relation(self) -> Optional[Relation]:
+        """Relation."""
+        if isinstance(self, MySQLAsyncReplicationOffer):
+            return self.model.get_relation(RELATION_OFFER)
+
+        return self.model.get_relation(RELATION_CONSUMER)
+
+    @property
+    def relation_data(self) -> Optional[RelationDataContent]:
+        """Relation data."""
+        if not self.relation:
             return
-        return relation.data[relation.app]
+        return self.relation.data[self.model.app]
 
-    def _on_promote_standby_cluster(self, event: ActionEvent) -> None:
+    @property
+    def remote_relation_data(self) -> Optional[RelationDataContent]:
+        """Remote relation data."""
+        if not self.relation or not self.relation.app:
+            return
+        return self.relation.data[self.relation.app]
+
+    def _on_promote_to_primary(self, event: ActionEvent) -> None:
         """Promote a standby cluster to primary."""
         if not self._charm.unit.is_leader():
             event.fail("Only the leader unit can promote a standby cluster")
@@ -189,7 +208,7 @@ def _on_fence_unfence_writes_action(self, event: ActionEvent) -> None:
         except MySQLFencingWritesError:
             event.fail("Failed to fence writes. Check logs for details")
 
-    def on_async_relation_broken(self, event):  # noqa: C901
+    def on_async_relation_broken(self, event: RelationBrokenEvent):  # noqa: C901
         """Handle the async relation being broken from either side."""
         # Remove the replica cluster, if this is the primary
 
@@ -213,7 +232,7 @@ def on_async_relation_broken(self, event):  # noqa: C901
                 logger.warning(
                     "Replica cluster not dissolved after relation broken by the primary cluster."
                     "\n\tThis happens when the primary cluster was removed prior to removing the async relation."
-                    "\n\tThis cluster can be promoted to primary with the `promote-standby-cluster` action."
+                    "\n\tThis cluster can be promoted to primary with the `promote-to-primary` action."
                 )
                 return
 
@@ -221,22 +240,26 @@ def on_async_relation_broken(self, event):  # noqa: C901
             # reset flag to allow instances rejoining the cluster
             self._charm.unit_peer_data["member-state"] = "waiting"
             del self._charm.unit_peer_data["unit-initialized"]
-            if self._charm.unit.is_leader():
-                self._charm.app.status = BlockedStatus("Recreate or rejoin cluster.")
-                logger.info(
-                    "\n\tThis is a replica cluster and will be dissolved.\n"
-                    "\tThe cluster can be recreated with the `recreate-cluster` action.\n"
-                    "\tAlternatively the cluster can be rejoined to the cluster set."
-                )
-                # reset the cluster node count flag
-                del self._charm.app_peer_data["units-added-to-cluster"]
-                # set flag to persist removed from cluster-set state
-                self._charm.app_peer_data["removed-from-cluster-set"] = "true"
+            if not self._charm.unit.is_leader():
+                # delay non leader to avoid `update_status` running before
+                # leader updates app peer data
+                sleep(10)
+                return
+            self._charm.app.status = BlockedStatus("Recreate or rejoin cluster.")
+            logger.info(
+                "\n\tThis is a replica cluster and will be dissolved.\n"
+                "\tThe cluster can be recreated with the `recreate-cluster` action.\n"
+                "\tAlternatively the cluster can be rejoined to the cluster set."
+            )
+            # reset the cluster node count flag
+            del self._charm.app_peer_data["units-added-to-cluster"]
+            # set flag to persist removed from cluster-set state
+            self._charm.app_peer_data["removed-from-cluster-set"] = "true"
 
         elif self.role.cluster_role == "primary":
             if self._charm.unit.is_leader():
                 # only leader units can remove replica clusters
-                remote_data = self.get_remote_relation_data(event.relation) or {}
+                remote_data = event.relation.data.get(event.relation.app) or {}
                 if cluster_name := remote_data.get("cluster-name"):
                     if self._charm._mysql.is_cluster_in_cluster_set(cluster_name):
                         self._charm.unit.status = MaintenanceStatus("Removing replica cluster")
@@ -264,6 +287,10 @@ def on_async_relation_broken(self, event):  # noqa: C901
                 self._charm.unit_peer_data["member-state"] = "unknown"
             self._charm._on_update_status(None)
 
+        if self._charm.app_peer_data.get("async-ready"):
+            # if set reset async-ready flag
+            del self._charm.app_peer_data["async-ready"]
+
     def _on_rejoin_cluster_action(self, event: ActionEvent) -> None:
         """Rejoin cluster to cluster set action handler."""
         cluster = event.params.get("cluster-name")
@@ -297,57 +324,53 @@ def _on_rejoin_cluster_action(self, event: ActionEvent) -> None:
             logger.error(message)
 
 
-class MySQLAsyncReplicationPrimary(MySQLAsyncReplication):
+class MySQLAsyncReplicationOffer(MySQLAsyncReplication):
     """MySQL async replication primary side.
 
     Implements the setup phase of the async replication for the primary side.
     """
 
     def __init__(self, charm: "MySQLOperatorCharm"):
-        super().__init__(charm, PRIMARY_RELATION)
+        super().__init__(charm, RELATION_OFFER)
 
         # Actions observed only on the primary class to avoid duplicated execution
         # promotion action since both classes are always instantiated
         self.framework.observe(
-            self._charm.on.promote_standby_cluster_action, self._on_promote_standby_cluster
+            self._charm.on.promote_to_primary_action, self._on_promote_to_primary
         )
 
-        # fence writes action
-        self.framework.observe(
-            self._charm.on.fence_writes_action, self._on_fence_unfence_writes_action
-        )
-        # unfence writes action
-        self.framework.observe(
-            self._charm.on.unfence_writes_action, self._on_fence_unfence_writes_action
-        )
         # rejoin invalidated cluster action
         self.framework.observe(
             self._charm.on.rejoin_cluster_action, self._on_rejoin_cluster_action
         )
 
+        # promote offer side as primary
+        self.framework.observe(
+            self._charm.on.create_replication_action, self._on_create_replication
+        )
+
         self.framework.observe(
-            self._charm.on[PRIMARY_RELATION].relation_created, self._on_primary_created
+            self._charm.on[RELATION_OFFER].relation_created, self._on_offer_created
         )
         self.framework.observe(
-            self._charm.on[PRIMARY_RELATION].relation_changed,
-            self._on_primary_relation_changed,
+            self._charm.on[RELATION_OFFER].relation_changed,
+            self._on_offer_relation_changed,
         )
 
-    def get_relation(self, relation_id: int) -> Optional[Relation]:
-        """Return the relation."""
-        return self.model.get_relation(PRIMARY_RELATION, relation_id)
+        self.framework.observe(
+            self._charm.on[RELATION_OFFER].relation_broken, self._on_offer_relation_broken
+        )
 
-    def get_local_relation_data(self, relation: Relation) -> Optional[RelationDataContent]:
-        """Local data."""
-        return relation.data[self.model.app]
+        self.framework.observe(self._charm.on.secret_changed, self._on_secret_change)
 
-    def get_state(self, relation: Relation) -> Optional[States]:
+    @property
+    def state(self) -> Optional[States]:
         """State of the relation, on primary side."""
-        if not relation:
+        if not self.relation:
             return States.UNINITIALIZED
 
-        local_data = self.get_local_relation_data(relation)
-        remote_data = self.get_remote_relation_data(relation) or {}
+        local_data = self.relation_data
+        remote_data = self.remote_relation_data or {}
 
         if not local_data:
             return States.UNINITIALIZED
@@ -377,82 +400,152 @@ def idle(self) -> bool:
             # non leader units are always idle
             return True
 
-        for relation in self.model.relations[PRIMARY_RELATION]:
-            if self.get_state(relation) not in [States.READY, States.UNINITIALIZED]:
-                return False
+        if self._charm.app_peer_data.get("async-ready") == "true":
+            # transitional state between relation created and setup_action
+            return False
+
+        if self.state not in [States.READY, States.UNINITIALIZED]:
+            return False
         return True
 
+    @property
+    def secret(self) -> Optional[Secret]:
+        """Return the async replication secret."""
+        if not self.relation:
+            return
+        if secret_id := self.relation_data.get("secret-id"):
+            return self._charm.model.get_secret(id=secret_id)
+
     def _get_secret(self) -> Secret:
         """Return async replication necessary secrets."""
-        try:
-            # Avoid recreating the secret
-            secret = self._charm.model.get_secret(label=SECRET_LABEL)
-            if not secret.id:
-                # workaround for the secret id not being set with model uuid
-                secret._id = f"secret://{self.model.uuid}/{secret.get_info().id.split(':')[1]}"
-            return secret
-        except SecretNotFoundError:
-            pass
-
         app_secret = self._charm.model.get_secret(label=f"{PEER}.{self.model.app.name}.app")
         content = app_secret.peek_content()
         # filter out unnecessary secrets
         shared_content = dict(filter(lambda x: "password" in x[0], content.items()))
 
-        return self._charm.model.app.add_secret(content=shared_content, label=SECRET_LABEL)
+        return self._charm.model.app.add_secret(content=shared_content)
+
+    def _on_create_replication(self, event: ActionEvent):
+        """Promote the offer side to primary on initial setup."""
+        if not self._charm.app_peer_data.get("async-ready") == "true":
+            event.fail("Relation created but not ready")
+            return
+
+        if self.role.relation_side == RELATION_CONSUMER:
+            # given that only the offer side of the relation can
+            # grant secret permissions for CMR relations, we
+            # limit the primary setup to it
+            event.fail("Only offer side can be setup as primary cluster")
+            return
 
-    def _on_primary_created(self, event):
-        """Validate relations and share credentials with replica cluster."""
         if not self._charm.unit.is_leader():
+            event.fail("Only the leader unit can promote a cluster")
             return
 
-        if not self._charm.unit_initialized:
-            logger.debug("Unit not initialized, deferring event")
-            event.defer()
+        if not self._charm.cluster_initialized:
+            event.fail("Wait until cluster is initialized")
             return
 
-        if self._charm._mysql.is_cluster_replica():
-            logger.error(
-                f"This is a replica cluster, cannot be related as {PRIMARY_RELATION}. Remove relation."
-            )
-            self._charm.unit.status = BlockedStatus(
-                f"This is a replica cluster. Unrelate from the {PRIMARY_RELATION} relation"
-            )
-            event.relation.data[self.model.app]["is-replica"] = "true"
+        if not self.relation:
+            event.fail(f"{RELATION_OFFER} relation not found")
+            return
+
+        if self.relation_data.get("secret-id"):
+            event.fail("Action already run")
             return
 
-        self._charm.app.status = MaintenanceStatus("Setting up async replication")
-        logger.info("Granting secrets access to async replication relation")
+        self._charm.app.status = MaintenanceStatus("Setting up replication")
+        self._charm.unit.status = MaintenanceStatus("Sharing credentials with replica cluster")
+        logger.info("Granting secrets access to replication relation")
         secret = self._get_secret()
-        secret_id = secret.id
-        secret.grant(event.relation)
+        secret_id = secret.id or ""
+        secret.grant(self.relation)
 
         # get workload version
-        version = self._charm._mysql.get_mysql_version()
+        version = self._charm._mysql.get_mysql_version() or "Unset"
 
         logger.debug(f"Sharing {secret_id=} with replica cluster")
         # Set variables for credential sync and validations
-        event.relation.data[self.model.app].update(
+        self.relation_data.update(  # pyright: ignore[reportCallIssue]
             {
                 "secret-id": secret_id,
                 "cluster-name": self.cluster_name,
                 "mysql-version": version,
+                "replication-name": event.params.get("name", "default"),
+            }
+        )
+        # reset async-ready flag set on relation created
+        del self._charm.app_peer_data["async-ready"]
+
+    def _on_offer_created(self, event: RelationCreatedEvent):
+        """Validate relations and share credentials with replica cluster."""
+        if not self._charm.unit.is_leader():
+            return
+
+        if (
+            isinstance(self._charm.app.status, BlockedStatus)
+            and self._charm.app_peer_data.get("removed-from-cluster-set") == "true"
+        ):
+            # Test for a broken relation on the primary side
+            logger.error(
+                (
+                    "Cannot setup async relation with primary cluster in blocked/read-only state\n"
+                    "Remove the relation."
+                )
+            )
+            message = f"Cluster is in a blocked state. Remove {RELATION_OFFER} relation"
+            self._charm.unit.status = BlockedStatus(message)
+            self._charm.app.status = BlockedStatus(message)
+
+        if not self.model.get_relation(RELATION_OFFER):
+            # safeguard against a deferred event a previous relation.
+            logger.error(
+                (
+                    "Relation created running against removed relation.\n"
+                    f"Remove {RELATION_OFFER} relation and retry."
+                )
+            )
+            self._charm.unit.status = BlockedStatus(f"Remove {RELATION_OFFER} relation and retry")
+            return
+
+        if not self._charm.cluster_initialized:
+            logger.info("Cluster not initialized, deferring event")
+            event.defer()
+            return
+
+        if self._charm._mysql.is_cluster_replica():
+            logger.error(
+                f"This is a replica cluster, cannot be related as {RELATION_OFFER}. Remove relation."
+            )
+            self._charm.unit.status = BlockedStatus(
+                f"This is a replica cluster. Unrelate from the {RELATION_OFFER} relation"
+            )
+            event.relation.data[self.model.app]["is-replica"] = "true"
+            return
+
+        self.relation_data.update(  # pyright: ignore[reportCallIssue]
+            {
                 "cluster-set-name": self.cluster_set_name,
             }
         )
+        # sets ok flag
+        self._charm.app_peer_data["async-ready"] = "true"
+        message = "Ready to create replication"
+        self._charm.unit.status = BlockedStatus(message)
+        self._charm.app.status = BlockedStatus(message)
 
-    def _on_primary_relation_changed(self, event):
+    def _on_offer_relation_changed(self, event):
         """Handle the async_primary relation being changed."""
         if not self._charm.unit.is_leader():
             return
 
-        state = self.get_state(event.relation)
+        state = self.state
 
         if state == States.INITIALIZING:
             # Add replica cluster primary node
             logger.info("Creating replica cluster primary node")
             self._charm.unit.status = MaintenanceStatus("Adding replica cluster")
-            remote_data = self.get_remote_relation_data(event.relation) or {}
+            remote_data = self.remote_relation_data or {}
 
             cluster = remote_data["cluster-name"]
             endpoint = remote_data["endpoint"]
@@ -481,53 +574,67 @@ def _on_primary_relation_changed(self, event):
             # Recover replica cluster
             self._charm.unit.status = MaintenanceStatus("Replica cluster in recovery")
 
+    def _on_offer_relation_broken(self, event: RelationBrokenEvent):
+        """Handle the async_primary relation being broken."""
+        if self._charm.unit.is_leader():
+            # remove relation secret by id
+            if secret_id := event.relation.data[self.model.app].get("secret-id"):
+                logger.debug("Removing replication secret")
+                secret = self._charm.model.get_secret(id=secret_id)
+                secret.remove_all_revisions()
+            else:
+                logger.debug("Secret-id not set, skipping removal")
+
+    def _on_secret_change(self, event: SecretChangedEvent):
+        """Propagates the secret being changed."""
+        if not self._charm.unit.is_leader():
+            return
+
+        if not self.relation:
+            return
+
+        if self.state != States.READY:
+            # skip secret propagation on setup phase
+            return
+
+        if event.secret.label != f"{PEER}.{self.model.app.name}.app":
+            # skip if secret is not main application secret
+            return
+
+        logger.debug("Updating secret on relation")
+        self.secret.set_content(event.secret.peek_content())
+
 
-class MySQLAsyncReplicationReplica(MySQLAsyncReplication):
+class MySQLAsyncReplicationConsumer(MySQLAsyncReplication):
     """MySQL async replication replica side.
 
     Implements the setup phase of the async replication for the replica side.
     """
 
     def __init__(self, charm: "MySQLOperatorCharm"):
-        super().__init__(charm, REPLICA_RELATION)
+        super().__init__(charm, RELATION_CONSUMER)
 
         # leader/primary
         self.framework.observe(
-            self._charm.on[REPLICA_RELATION].relation_created, self._on_replica_created
+            self._charm.on[RELATION_CONSUMER].relation_created, self._on_consumer_relation_created
         )
         self.framework.observe(
-            self._charm.on[REPLICA_RELATION].relation_changed, self._on_replica_changed
+            self._charm.on[RELATION_CONSUMER].relation_changed, self._on_consumer_changed
         )
         # non-leader/secondaries
         self.framework.observe(
-            self._charm.on[REPLICA_RELATION].relation_created,
-            self._on_replica_non_leader_created,
+            self._charm.on[RELATION_CONSUMER].relation_created,
+            self._on_consumer_non_leader_created,
         )
         self.framework.observe(
-            self._charm.on[REPLICA_RELATION].relation_changed,
-            self._on_replica_non_leader_changed,
+            self._charm.on[RELATION_CONSUMER].relation_changed,
+            self._on_consumer_non_leader_changed,
         )
-
-    @property
-    def relation(self) -> Optional[Relation]:
-        """Relation."""
-        return self.model.get_relation(REPLICA_RELATION)
-
-    @property
-    def relation_data(self) -> RelationDataContent:
-        """Relation data."""
-        return self.relation.data[self.model.app]
-
-    @property
-    def remote_relation_data(self) -> Optional[RelationDataContent]:
-        """Relation data."""
-        if not self.relation.app:
-            return
-        return self.relation.data[self.relation.app]
+        self.framework.observe(self._charm.on.secret_changed, self._on_secret_change)
 
     @property
     def state(self) -> Optional[States]:
-        """State of the relation, on replica side."""
+        """State of the relation, on consumer side."""
         if not self.relation:
             return None
 
@@ -591,7 +698,7 @@ def _check_version(self) -> bool:
     def _obtain_secret(self) -> Secret:
         """Get secret from primary cluster."""
         secret_id = self.remote_relation_data.get("secret-id")
-        return self._charm.model.get_secret(id=secret_id, label=SECRET_LABEL)
+        return self._charm.model.get_secret(id=secret_id)
 
     def _async_replication_credentials(self) -> dict[str, str]:
         """Get async replication credentials from primary cluster."""
@@ -608,14 +715,16 @@ def _get_endpoint(self) -> str:
         # using unit informed address (fqdn or ip)
         return self._charm.unit_address
 
-    def _on_replica_created(self, event):
+    def _on_consumer_relation_created(self, event):
         """Handle the async_replica relation being created on the leader unit."""
         if not self._charm.unit.is_leader():
             return
         if not self._charm.unit_initialized and not self.returning_cluster:
             # avoid running too early for non returning clusters
-            logger.debug("Unit not initialized, deferring event")
-            event.defer()
+            self._charm.unit.status = BlockedStatus(
+                "Wait until unit is initialized before running create-replication on offer side"
+            )
+            self._charm.app.status = MaintenanceStatus("Setting up replication")
             return
         if self.returning_cluster:
             # flag set on prior async relation broken
@@ -641,10 +750,10 @@ def _on_replica_created(self, event):
                 self.relation_data["user-data-found"] = "true"
                 return
 
-        self._charm.app.status = MaintenanceStatus("Setting up async replication")
+        self._charm.app.status = MaintenanceStatus("Setting up replication")
         self._charm.unit.status = WaitingStatus("Awaiting sync data from primary cluster")
 
-    def _on_replica_changed(self, event):  # noqa: C901
+    def _on_consumer_changed(self, event):  # noqa: C901
         """Handle the async_replica relation being changed."""
         if not self._charm.unit.is_leader():
             return
@@ -682,7 +791,7 @@ def _on_replica_changed(self, event):  # noqa: C901
 
             logger.debug("Syncing credentials from primary cluster")
             self._charm.unit.status = MaintenanceStatus("Syncing credentials")
-            self._charm.app.status = MaintenanceStatus("Setting up async replication")
+            self._charm.app.status = MaintenanceStatus("Setting up replication")
 
             try:
                 credentials = self._async_replication_credentials()
@@ -717,14 +826,14 @@ def _on_replica_changed(self, event):  # noqa: C901
             # reset force rejoin-secondaries flag
             del self._charm.app_peer_data["rejoin-secondaries"]
 
-            if self.remote_relation_data["cluster-name"] == self.cluster_name:
+            if self.remote_relation_data["cluster-name"] == self.cluster_name:  # pyright: ignore
                 # this cluster need a new cluster name
                 logger.warning(
                     "Cluster name is the same as the primary cluster. Appending generated value"
                 )
-                self._charm.app_peer_data[
-                    "cluster-name"
-                ] = f"{self.cluster_name}{uuid.uuid4().hex[:4]}"
+                self._charm.app_peer_data["cluster-name"] = (
+                    f"{self.cluster_name}{uuid.uuid4().hex[:4]}"
+                )
 
             self._charm.unit.status = MaintenanceStatus("Populate endpoint")
 
@@ -768,17 +877,21 @@ def _on_replica_changed(self, event):  # noqa: C901
             self._charm.unit_peer_data["member-role"] = "primary"
             event.defer()
 
-    def _on_replica_non_leader_created(self, _):
-        """Handle the async_replica relation being created for secondaries/non-leader."""
+    def _on_consumer_non_leader_created(self, _):
+        """Handle the consumer relation being created for secondaries/non-leader."""
         # set waiting state to inhibit auto recovery, only when not already set
+        if self._charm.unit.is_leader():
+            return
         if not self._charm.unit_peer_data.get("member-state") == "waiting":
             self._charm.unit_peer_data["member-state"] = "waiting"
             self._charm.unit.status = WaitingStatus("waiting replica cluster be configured")
 
-    def _on_replica_non_leader_changed(self, _):
+    def _on_consumer_non_leader_changed(self, _):
         """Reset cluster secondaries to allow cluster rejoin after primary recovery."""
         # the replica state is initialized when the primary cluster finished
         # creating the replica cluster on this cluster primary/leader unit
+        if self._charm.unit.is_leader():
+            return
         if (
             self.replica_initialized
             or self._charm.app_peer_data.get("rejoin-secondaries") == "true"
@@ -789,3 +902,24 @@ def _on_replica_non_leader_changed(self, _):
             del self._charm.unit_peer_data["unit-initialized"]
             self._charm.unit_peer_data["member-state"] = "waiting"
             self._charm.unit.status = WaitingStatus("waiting to join the cluster")
+
+    def _on_secret_change(self, event: SecretChangedEvent):
+        """Propagates the secret being changed."""
+        if not self._charm.unit.is_leader():
+            return
+
+        if not self.relation:
+            return
+
+        if event.secret.id.removeprefix("secret:") not in self.remote_relation_data.get(
+            "secret-id", "dummy"
+        ):
+            logger.info("Secret not related to replication")
+            return
+
+        logger.debug("Propagating secret change from the relation offer")
+        credentials = self._async_replication_credentials()
+
+        self._charm.model.get_secret(label=f"{PEER}.{self.model.app.name}.app").set_content(
+            credentials
+        )
diff --git a/lib/charms/mysql/v0/mysql.py b/lib/charms/mysql/v0/mysql.py
index 1168dd871..833f6a912 100644
--- a/lib/charms/mysql/v0/mysql.py
+++ b/lib/charms/mysql/v0/mysql.py
@@ -79,7 +79,18 @@ def wait_until_mysql_connection(self) -> None:
 import time
 from abc import ABC, abstractmethod
 from pathlib import Path
-from typing import Any, Dict, Iterable, List, Literal, Optional, Tuple, Union, get_args
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    Iterable,
+    List,
+    Literal,
+    Optional,
+    Tuple,
+    Union,
+    get_args,
+)
 
 import ops
 from charms.data_platform_libs.v0.data_interfaces import DataPeerData, DataPeerUnitData
@@ -108,6 +119,9 @@ def wait_until_mysql_connection(self) -> None:
 
 logger = logging.getLogger(__name__)
 
+if TYPE_CHECKING:
+    from charms.mysql.v0.async_replication import MySQLAsyncReplicationOffer
+
 # The unique Charmhub library identifier, never change it
 LIBID = "8c1428f06b1b4ec8bf98b7d980a38a8c"
 
@@ -116,7 +130,7 @@ def wait_until_mysql_connection(self) -> None:
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 58
+LIBPATCH = 59
 
 UNIT_TEARDOWN_LOCKNAME = "unit-teardown"
 UNIT_ADD_LOCKNAME = "unit-add"
@@ -134,7 +148,7 @@ def wait_until_mysql_connection(self) -> None:
 
 APP_SCOPE = "app"
 UNIT_SCOPE = "unit"
-Scopes = Literal[APP_SCOPE, UNIT_SCOPE]
+Scopes = Literal["app", "unit"]
 
 
 class Error(Exception):
@@ -402,6 +416,8 @@ class MySQLCharmBase(CharmBase, ABC):
     K8s charms.
     """
 
+    replication_offer: "MySQLAsyncReplicationOffer"
+
     def __init__(self, *args):
         super().__init__(*args)
 
@@ -477,6 +493,10 @@ def _on_set_password(self, event: ActionEvent) -> None:
             event.fail("set-password action can only be run on the leader unit.")
             return
 
+        if self.replication_offer.role.relation_side != "replication-offer":
+            event.fail("Only offer side can change password when replications is enabled")
+            return
+
         username = event.params.get("username") or ROOT_USERNAME
 
         valid_usernames = {
@@ -548,6 +568,7 @@ def _recreate_cluster(self, event: ActionEvent) -> None:
         try:
             self.create_cluster()
             self.unit.status = ops.ActiveStatus(self.active_status_message)
+            self.app.status = ops.ActiveStatus()
         except (MySQLCreateClusterError, MySQLCreateClusterSetError) as e:
             logger.exception("Failed to recreate cluster")
             event.fail(str(e))
@@ -659,9 +680,9 @@ def active_status_message(self) -> str:
             if self._mysql.is_cluster_replica():
                 status = self._mysql.get_replica_cluster_status()
                 if status == "ok":
-                    return "Primary (standby)"
+                    return "Standby"
                 else:
-                    return f"Primary (standby, {status})"
+                    return f"Standby ({status})"
             elif self._mysql.is_cluster_writes_fenced():
                 return "Primary (fenced writes)"
             else:
@@ -2380,8 +2401,12 @@ def update_user_password(self, username: str, new_password: str, host: str = "%"
         """
         logger.debug(f"Updating password for {username}.")
 
+        # password is set on the global primary
+        if not (instance_address := self.get_cluster_set_global_primary_address()):
+            raise MySQLCheckUserExistenceError("No primary found")
+
         update_user_password_commands = (
-            f"shell.connect_to_primary('{self.server_config_user}:{self.server_config_password}@{self.instance_address}')",
+            f"shell.connect('{self.server_config_user}:{self.server_config_password}@{instance_address}')",
             f"session.run_sql(\"ALTER USER '{username}'@'{host}' IDENTIFIED BY '{new_password}';\")",
             'session.run_sql("FLUSH PRIVILEGES;")',
         )
@@ -3078,7 +3103,7 @@ def flush_mysql_logs(self, logs_type: Union[MySQLTextLogs, list[MySQLTextLogs]])
             'session.run_sql("SET sql_log_bin = 0")',
         ]
 
-        if type(logs_type) is list:
+        if isinstance(logs_type, list):
             flush_logs_commands.extend(
                 [f"session.run_sql('FLUSH {log.value}')" for log in logs_type]
             )
diff --git a/metadata.yaml b/metadata.yaml
index 0a15cf862..08c96de48 100644
--- a/metadata.yaml
+++ b/metadata.yaml
@@ -41,8 +41,9 @@ provides:
   cos-agent:
     interface: cos_agent
     limit: 1
-  async-primary:
-    interface: async_replication
+  replication-offer:
+    interface: mysql_async
+    limit: 1
 
 requires:
   certificates:
@@ -53,8 +54,8 @@ requires:
     interface: s3
     limit: 1
     optional: true
-  async-replica:
-    interface: async_replication
+  replication:
+    interface: mysql_async
     limit: 1
     optional: true
 
@@ -66,3 +67,10 @@ storage:
 
 assumes:
   - juju
+  - any-of:
+      - all-of:
+          - juju >= 2.9.44
+          - juju < 3
+      - all-of:
+          - juju >= 3.4.3
+          - juju < 4
diff --git a/poetry.lock b/poetry.lock
index 8380e1b89..26600933c 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
 
 [[package]]
 name = "allure-pytest"
@@ -123,33 +123,33 @@ typecheck = ["mypy"]
 
 [[package]]
 name = "black"
-version = "23.7.0"
+version = "24.4.2"
 description = "The uncompromising code formatter."
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "black-23.7.0-cp310-cp310-macosx_10_16_arm64.whl", hash = "sha256:5c4bc552ab52f6c1c506ccae05681fab58c3f72d59ae6e6639e8885e94fe2587"},
-    {file = "black-23.7.0-cp310-cp310-macosx_10_16_universal2.whl", hash = "sha256:552513d5cd5694590d7ef6f46e1767a4df9af168d449ff767b13b084c020e63f"},
-    {file = "black-23.7.0-cp310-cp310-macosx_10_16_x86_64.whl", hash = "sha256:86cee259349b4448adb4ef9b204bb4467aae74a386bce85d56ba4f5dc0da27be"},
-    {file = "black-23.7.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:501387a9edcb75d7ae8a4412bb8749900386eaef258f1aefab18adddea1936bc"},
-    {file = "black-23.7.0-cp310-cp310-win_amd64.whl", hash = "sha256:fb074d8b213749fa1d077d630db0d5f8cc3b2ae63587ad4116e8a436e9bbe995"},
-    {file = "black-23.7.0-cp311-cp311-macosx_10_16_arm64.whl", hash = "sha256:b5b0ee6d96b345a8b420100b7d71ebfdd19fab5e8301aff48ec270042cd40ac2"},
-    {file = "black-23.7.0-cp311-cp311-macosx_10_16_universal2.whl", hash = "sha256:893695a76b140881531062d48476ebe4a48f5d1e9388177e175d76234ca247cd"},
-    {file = "black-23.7.0-cp311-cp311-macosx_10_16_x86_64.whl", hash = "sha256:c333286dc3ddca6fdff74670b911cccedacb4ef0a60b34e491b8a67c833b343a"},
-    {file = "black-23.7.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:831d8f54c3a8c8cf55f64d0422ee875eecac26f5f649fb6c1df65316b67c8926"},
-    {file = "black-23.7.0-cp311-cp311-win_amd64.whl", hash = "sha256:7f3bf2dec7d541b4619b8ce526bda74a6b0bffc480a163fed32eb8b3c9aed8ad"},
-    {file = "black-23.7.0-cp38-cp38-macosx_10_16_arm64.whl", hash = "sha256:f9062af71c59c004cd519e2fb8f5d25d39e46d3af011b41ab43b9c74e27e236f"},
-    {file = "black-23.7.0-cp38-cp38-macosx_10_16_universal2.whl", hash = "sha256:01ede61aac8c154b55f35301fac3e730baf0c9cf8120f65a9cd61a81cfb4a0c3"},
-    {file = "black-23.7.0-cp38-cp38-macosx_10_16_x86_64.whl", hash = "sha256:327a8c2550ddc573b51e2c352adb88143464bb9d92c10416feb86b0f5aee5ff6"},
-    {file = "black-23.7.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6d1c6022b86f83b632d06f2b02774134def5d4d4f1dac8bef16d90cda18ba28a"},
-    {file = "black-23.7.0-cp38-cp38-win_amd64.whl", hash = "sha256:27eb7a0c71604d5de083757fbdb245b1a4fae60e9596514c6ec497eb63f95320"},
-    {file = "black-23.7.0-cp39-cp39-macosx_10_16_arm64.whl", hash = "sha256:8417dbd2f57b5701492cd46edcecc4f9208dc75529bcf76c514864e48da867d9"},
-    {file = "black-23.7.0-cp39-cp39-macosx_10_16_universal2.whl", hash = "sha256:47e56d83aad53ca140da0af87678fb38e44fd6bc0af71eebab2d1f59b1acf1d3"},
-    {file = "black-23.7.0-cp39-cp39-macosx_10_16_x86_64.whl", hash = "sha256:25cc308838fe71f7065df53aedd20327969d05671bac95b38fdf37ebe70ac087"},
-    {file = "black-23.7.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:642496b675095d423f9b8448243336f8ec71c9d4d57ec17bf795b67f08132a91"},
-    {file = "black-23.7.0-cp39-cp39-win_amd64.whl", hash = "sha256:ad0014efc7acf0bd745792bd0d8857413652979200ab924fbf239062adc12491"},
-    {file = "black-23.7.0-py3-none-any.whl", hash = "sha256:9fd59d418c60c0348505f2ddf9609c1e1de8e7493eab96198fc89d9f865e7a96"},
-    {file = "black-23.7.0.tar.gz", hash = "sha256:022a582720b0d9480ed82576c920a8c1dde97cc38ff11d8d8859b3bd6ca9eedb"},
+    {file = "black-24.4.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:dd1b5a14e417189db4c7b64a6540f31730713d173f0b63e55fabd52d61d8fdce"},
+    {file = "black-24.4.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:8e537d281831ad0e71007dcdcbe50a71470b978c453fa41ce77186bbe0ed6021"},
+    {file = "black-24.4.2-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:eaea3008c281f1038edb473c1aa8ed8143a5535ff18f978a318f10302b254063"},
+    {file = "black-24.4.2-cp310-cp310-win_amd64.whl", hash = "sha256:7768a0dbf16a39aa5e9a3ded568bb545c8c2727396d063bbaf847df05b08cd96"},
+    {file = "black-24.4.2-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:257d724c2c9b1660f353b36c802ccece186a30accc7742c176d29c146df6e474"},
+    {file = "black-24.4.2-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:bdde6f877a18f24844e381d45e9947a49e97933573ac9d4345399be37621e26c"},
+    {file = "black-24.4.2-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e151054aa00bad1f4e1f04919542885f89f5f7d086b8a59e5000e6c616896ffb"},
+    {file = "black-24.4.2-cp311-cp311-win_amd64.whl", hash = "sha256:7e122b1c4fb252fd85df3ca93578732b4749d9be076593076ef4d07a0233c3e1"},
+    {file = "black-24.4.2-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:accf49e151c8ed2c0cdc528691838afd217c50412534e876a19270fea1e28e2d"},
+    {file = "black-24.4.2-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:88c57dc656038f1ab9f92b3eb5335ee9b021412feaa46330d5eba4e51fe49b04"},
+    {file = "black-24.4.2-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:be8bef99eb46d5021bf053114442914baeb3649a89dc5f3a555c88737e5e98fc"},
+    {file = "black-24.4.2-cp312-cp312-win_amd64.whl", hash = "sha256:415e686e87dbbe6f4cd5ef0fbf764af7b89f9057b97c908742b6008cc554b9c0"},
+    {file = "black-24.4.2-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:bf10f7310db693bb62692609b397e8d67257c55f949abde4c67f9cc574492cc7"},
+    {file = "black-24.4.2-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:98e123f1d5cfd42f886624d84464f7756f60ff6eab89ae845210631714f6db94"},
+    {file = "black-24.4.2-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:48a85f2cb5e6799a9ef05347b476cce6c182d6c71ee36925a6c194d074336ef8"},
+    {file = "black-24.4.2-cp38-cp38-win_amd64.whl", hash = "sha256:b1530ae42e9d6d5b670a34db49a94115a64596bc77710b1d05e9801e62ca0a7c"},
+    {file = "black-24.4.2-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:37aae07b029fa0174d39daf02748b379399b909652a806e5708199bd93899da1"},
+    {file = "black-24.4.2-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:da33a1a5e49c4122ccdfd56cd021ff1ebc4a1ec4e2d01594fef9b6f267a9e741"},
+    {file = "black-24.4.2-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ef703f83fc32e131e9bcc0a5094cfe85599e7109f896fe8bc96cc402f3eb4b6e"},
+    {file = "black-24.4.2-cp39-cp39-win_amd64.whl", hash = "sha256:b9176b9832e84308818a99a561e90aa479e73c523b3f77afd07913380ae2eab7"},
+    {file = "black-24.4.2-py3-none-any.whl", hash = "sha256:d36ed1124bb81b32f8614555b34cc4259c3fbc7eec17870e8ff8ded335b58d8c"},
+    {file = "black-24.4.2.tar.gz", hash = "sha256:c872b53057f000085da66a19c55d68f6f8ddcac2642392ad3a355878406fbd4d"},
 ]
 
 [package.dependencies]
@@ -159,10 +159,11 @@ packaging = ">=22.0"
 pathspec = ">=0.9.0"
 platformdirs = ">=2"
 tomli = {version = ">=1.1.0", markers = "python_version < \"3.11\""}
+typing-extensions = {version = ">=4.0.1", markers = "python_version < \"3.11\""}
 
 [package.extras]
 colorama = ["colorama (>=0.4.3)"]
-d = ["aiohttp (>=3.7.4)"]
+d = ["aiohttp (>=3.7.4)", "aiohttp (>=3.7.4,!=3.9.0)"]
 jupyter = ["ipython (>=7.8.0)", "tokenize-rt (>=3.2.0)"]
 uvloop = ["uvloop (>=0.15.2)"]
 
@@ -614,19 +615,19 @@ tests = ["asttokens", "littleutils", "pytest", "rich"]
 
 [[package]]
 name = "flake8"
-version = "6.0.0"
+version = "7.0.0"
 description = "the modular source code checker: pep8 pyflakes and co"
 optional = false
 python-versions = ">=3.8.1"
 files = [
-    {file = "flake8-6.0.0-py2.py3-none-any.whl", hash = "sha256:3833794e27ff64ea4e9cf5d410082a8b97ff1a06c16aa3d2027339cd0f1195c7"},
-    {file = "flake8-6.0.0.tar.gz", hash = "sha256:c61007e76655af75e6785a931f452915b371dc48f56efd765247c8fe68f2b181"},
+    {file = "flake8-7.0.0-py2.py3-none-any.whl", hash = "sha256:a6dfbb75e03252917f2473ea9653f7cd799c3064e54d4c8140044c5c065f53c3"},
+    {file = "flake8-7.0.0.tar.gz", hash = "sha256:33f96621059e65eec474169085dc92bf26e7b2d47366b70be2f67ab80dc25132"},
 ]
 
 [package.dependencies]
 mccabe = ">=0.7.0,<0.8.0"
-pycodestyle = ">=2.10.0,<2.11.0"
-pyflakes = ">=3.0.0,<3.1.0"
+pycodestyle = ">=2.11.0,<2.12.0"
+pyflakes = ">=3.2.0,<3.3.0"
 
 [[package]]
 name = "flake8-builtins"
@@ -1374,13 +1375,13 @@ pyasn1 = ">=0.4.6,<0.6.0"
 
 [[package]]
 name = "pycodestyle"
-version = "2.10.0"
+version = "2.11.1"
 description = "Python style guide checker"
 optional = false
-python-versions = ">=3.6"
+python-versions = ">=3.8"
 files = [
-    {file = "pycodestyle-2.10.0-py2.py3-none-any.whl", hash = "sha256:8a4eaf0d0495c7395bdab3589ac2db602797d76207242c17d470186815706610"},
-    {file = "pycodestyle-2.10.0.tar.gz", hash = "sha256:347187bdb476329d98f695c213d7295a846d1152ff4fe9bacb8a9590b8ee7053"},
+    {file = "pycodestyle-2.11.1-py2.py3-none-any.whl", hash = "sha256:44fe31000b2d866f2e41841b18528a505fbd7fef9017b04eff4e2648a0fadc67"},
+    {file = "pycodestyle-2.11.1.tar.gz", hash = "sha256:41ba0e7afc9752dfb53ced5489e89f8186be00e599e712660695b7a75ff2663f"},
 ]
 
 [[package]]
@@ -1465,13 +1466,13 @@ toml = ["tomli (>=1.2.3)"]
 
 [[package]]
 name = "pyflakes"
-version = "3.0.1"
+version = "3.2.0"
 description = "passive checker of Python programs"
 optional = false
-python-versions = ">=3.6"
+python-versions = ">=3.8"
 files = [
-    {file = "pyflakes-3.0.1-py2.py3-none-any.whl", hash = "sha256:ec55bf7fe21fff7f1ad2f7da62363d749e2a470500eab1b555334b67aa1ef8cf"},
-    {file = "pyflakes-3.0.1.tar.gz", hash = "sha256:ec8b276a6b60bd80defed25add7e439881c19e64850afd9b346283d4165fd0fd"},
+    {file = "pyflakes-3.2.0-py2.py3-none-any.whl", hash = "sha256:84b5be138a2dfbb40689ca07e2152deb896a65c3a3e24c251c5c62489568074a"},
+    {file = "pyflakes-3.2.0.tar.gz", hash = "sha256:1c61603ff154621fb2a9172037d84dca3500def8c8b630657d1701f026f8af3f"},
 ]
 
 [[package]]
@@ -1560,17 +1561,17 @@ test = ["pretend", "pytest (>=3.0.1)", "pytest-rerunfailures"]
 
 [[package]]
 name = "pyproject-flake8"
-version = "6.0.0.post1"
+version = "7.0.0"
 description = "pyproject-flake8 (`pflake8`), a monkey patching wrapper to connect flake8 with pyproject.toml configuration"
 optional = false
 python-versions = ">=3.8.1"
 files = [
-    {file = "pyproject-flake8-6.0.0.post1.tar.gz", hash = "sha256:d43421caca0ef8a672874405fe63c722b0333e3c22c41648c6df60f21bab2b6b"},
-    {file = "pyproject_flake8-6.0.0.post1-py3-none-any.whl", hash = "sha256:bdc7ca9b967b9724983903489b8943b72c668178fb69f03e8774ec74f6a13782"},
+    {file = "pyproject_flake8-7.0.0-py3-none-any.whl", hash = "sha256:611e91b49916e6d0685f88423ad4baff490888278a258975403c0dee6eb6072e"},
+    {file = "pyproject_flake8-7.0.0.tar.gz", hash = "sha256:5b953592336bc04d86e8942fdca1014256044a3445c8b6ca9467d08636749158"},
 ]
 
 [package.dependencies]
-flake8 = "6.0.0"
+flake8 = "7.0.0"
 tomli = {version = "*", markers = "python_version < \"3.11\""}
 
 [[package]]
@@ -2285,4 +2286,4 @@ files = [
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.10"
-content-hash = "099499829dacf52d7a1bb57bb08d1c85204c062c1536e319a5319d7e797b6962"
+content-hash = "44041cdc89d8ebd562c87c696461b8404358cbf16db0beaac2048f79add80a0a"
diff --git a/pyproject.toml b/pyproject.toml
index 8253e17f1..6c02842d5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -37,20 +37,20 @@ jsonschema = "*"
 optional = true
 
 [tool.poetry.group.format.dependencies]
-black = "^23.7.0"
+black = "^24.0.0"
 isort = "^5.12.0"
 
 [tool.poetry.group.lint]
 optional = true
 
 [tool.poetry.group.lint.dependencies]
-black = "^23.7.0"
+black = "^24.0.0"
 isort = "^5.12.0"
-flake8 = "^6.0.0"
+flake8 = "^7.0.0"
 flake8-docstrings = "^1.7.0"
 flake8-copyright = "^0.2.4"
 flake8-builtins = "^2.1.0"
-pyproject-flake8 = "^6.0.0.post1"
+pyproject-flake8 = "^7.0.0"
 pep8-naming = "^0.13.3"
 codespell = "^2.2.5"
 shellcheck-py = "^0.9.0.5"
diff --git a/src/charm.py b/src/charm.py
index b6da3d0bb..7042736f4 100755
--- a/src/charm.py
+++ b/src/charm.py
@@ -16,8 +16,8 @@
 from charms.data_platform_libs.v0.s3 import S3Requirer
 from charms.grafana_agent.v0.cos_agent import COSAgentProvider
 from charms.mysql.v0.async_replication import (
-    MySQLAsyncReplicationPrimary,
-    MySQLAsyncReplicationReplica,
+    MySQLAsyncReplicationConsumer,
+    MySQLAsyncReplicationOffer,
 )
 from charms.mysql.v0.backups import MySQLBackups
 from charms.mysql.v0.mysql import (
@@ -167,8 +167,8 @@ def __init__(self, *args):
         self.restart = RollingOpsManager(self, relation="restart", callback=self._restart)
 
         self.mysql_logs = MySQLLogs(self)
-        self.async_primary = MySQLAsyncReplicationPrimary(self)
-        self.async_replica = MySQLAsyncReplicationReplica(self)
+        self.replication_offer = MySQLAsyncReplicationOffer(self)
+        self.replication_consumer = MySQLAsyncReplicationConsumer(self)
 
     # =======================
     #  Charm Lifecycle Hooks
@@ -446,7 +446,7 @@ def _on_update_status(self, _) -> None:  # noqa: C901
             logger.debug("skip status update while upgrading")
             return
 
-        if not (self.async_primary.idle and self.async_replica.idle):
+        if not (self.replication_offer.idle and self.replication_consumer.idle):
             # avoid changing status while in async replication
             logger.debug("skip status update while setting up async replication")
             return
diff --git a/tests/integration/helpers.py b/tests/integration/helpers.py
index dbc371f6f..6cd75ca06 100644
--- a/tests/integration/helpers.py
+++ b/tests/integration/helpers.py
@@ -723,7 +723,7 @@ async def write_random_chars_to_test_table(ops_test: OpsTest, primary_unit: Unit
         "CREATE TABLE test.data_replication_table (id varchar(40), primary key(id))",
         (
             "INSERT INTO test.data_replication_table"
-            f" VALUES ('{(random_chars:=generate_random_string(40))}')"
+            f" VALUES ('{(random_chars := generate_random_string(40))}')"
         ),
     ]
 
diff --git a/tests/integration/high_availability/test_async_replication.py b/tests/integration/high_availability/test_async_replication.py
index 735b769bf..98466cd4b 100644
--- a/tests/integration/high_availability/test_async_replication.py
+++ b/tests/integration/high_availability/test_async_replication.py
@@ -110,25 +110,59 @@ async def test_build_and_deploy(
 async def test_async_relate(first_model: Model, second_model: Model) -> None:
     """Relate the two mysql clusters."""
     logger.info("Creating offers in first model")
-    await first_model.create_offer(f"{MYSQL_APP1}:async-primary")
+    await first_model.create_offer(f"{MYSQL_APP1}:replication-offer")
 
     logger.info("Consume offer in second model")
     await second_model.consume(endpoint=f"admin/{first_model.info.name}.{MYSQL_APP1}")
 
     logger.info("Relating the two mysql clusters")
-    await second_model.integrate(f"{MYSQL_APP1}", f"{MYSQL_APP2}:async-replica")
+    await second_model.integrate(f"{MYSQL_APP1}", f"{MYSQL_APP2}:replication")
+
+    logger.info("Waiting for the applications to settle")
+    await gather(
+        first_model.block_until(
+            lambda: any(
+                unit.workload_status == "blocked"
+                for unit in first_model.applications[MYSQL_APP1].units
+            ),
+            timeout=5 * MINUTE,
+        ),
+        second_model.block_until(
+            lambda: all(
+                unit.workload_status == "waiting"
+                for unit in second_model.applications[MYSQL_APP2].units
+            ),
+            timeout=5 * MINUTE,
+        ),
+    )
+
+
+@juju3
+@pytest.mark.abort_on_fail
+@pytest.mark.group(1)
+async def test_create_replication(first_model: Model, second_model: Model) -> None:
+    """Run the create replication and wait for the applications to settle."""
+    logger.info("Running create replication action")
+    leader_unit = await get_leader_unit(None, MYSQL_APP1, first_model)
+    assert leader_unit is not None, "No leader unit found"
+
+    await juju_.run_action(
+        leader_unit,
+        "create-replication",
+        **{"--wait": "5m"},
+    )
 
     logger.info("Waiting for the applications to settle")
     await gather(
         first_model.wait_for_idle(
             apps=[MYSQL_APP1],
             status="active",
-            timeout=10 * MINUTE,
+            timeout=5 * MINUTE,
         ),
         second_model.wait_for_idle(
             apps=[MYSQL_APP2],
             status="active",
-            timeout=10 * MINUTE,
+            timeout=5 * MINUTE,
         ),
     )
 
@@ -198,7 +232,7 @@ async def test_standby_promotion(
     cluster_set_name = relation_data[0]["application-data"]["cluster-set-domain-name"]
     logger.info("Promoting standby cluster to primary")
     await juju_.run_action(
-        leader_unit, "promote-standby-cluster", **{"cluster-set-name": cluster_set_name}
+        leader_unit, "promote-to-primary", **{"cluster-set-name": cluster_set_name}
     )
 
     results = await get_max_written_value(first_model, second_model)
@@ -231,7 +265,7 @@ async def test_failover(ops_test: OpsTest, first_model: Model, second_model: Mod
     cluster_set_name = relation_data[0]["application-data"]["cluster-set-domain-name"]
     await juju_.run_action(
         leader_unit,
-        "promote-standby-cluster",
+        "promote-to-primary",
         **{"--wait": "5m", "cluster-set-name": cluster_set_name, "force": True},
     )
 
@@ -292,7 +326,7 @@ async def test_remove_relation_and_relate(
 
     logger.info("Remove async relation")
     await second_model.applications[MYSQL_APP2].remove_relation(
-        f"{MYSQL_APP2}:async-replica", MYSQL_APP1
+        f"{MYSQL_APP2}:replication", MYSQL_APP1
     )
 
     second_model_units = second_model.applications[MYSQL_APP2].units
@@ -310,7 +344,26 @@ async def test_remove_relation_and_relate(
     )
 
     logger.info("Re relating the two mysql clusters")
-    await second_model.integrate(f"{MYSQL_APP1}", f"{MYSQL_APP2}:async-replica")
+    await second_model.integrate(f"{MYSQL_APP1}", f"{MYSQL_APP2}:replication")
+
+    logger.info("Waiting for the applications to settle")
+    await first_model.block_until(
+        lambda: any(
+            unit.workload_status == "blocked"
+            for unit in first_model.applications[MYSQL_APP1].units
+        ),
+        timeout=5 * MINUTE,
+    )
+
+    logger.info("Running create replication action")
+    leader_unit = await get_leader_unit(None, MYSQL_APP1, first_model)
+    assert leader_unit is not None, "No leader unit found"
+
+    await juju_.run_action(
+        leader_unit,
+        "create-replication",
+        **{"--wait": "5m"},
+    )
 
     logger.info("Waiting for the applications to settle")
     await gather(
diff --git a/tests/integration/high_availability/test_upgrade.py b/tests/integration/high_availability/test_upgrade.py
index 5b828a1bc..06c6b89d3 100644
--- a/tests/integration/high_availability/test_upgrade.py
+++ b/tests/integration/high_availability/test_upgrade.py
@@ -183,7 +183,7 @@ async def inject_dependency_fault(
 
     loaded_dependency_dict = json.loads(relation_data[0]["application-data"]["dependencies"])
     loaded_dependency_dict["charm"]["upgrade_supported"] = f">{current_charm_version}"
-    loaded_dependency_dict["charm"]["version"] = f"{int(current_charm_version)+1}"
+    loaded_dependency_dict["charm"]["version"] = f"{int(current_charm_version) + 1}"
 
     # Overwrite dependency.json with incompatible version
     with zipfile.ZipFile(charm_file, mode="a") as charm_zip:
diff --git a/tests/unit/test_async_replication.py b/tests/unit/test_async_replication.py
index 5660d6782..61b437cb2 100644
--- a/tests/unit/test_async_replication.py
+++ b/tests/unit/test_async_replication.py
@@ -6,8 +6,8 @@
 
 import pytest
 from charms.mysql.v0.async_replication import (
-    PRIMARY_RELATION,
-    REPLICA_RELATION,
+    RELATION_CONSUMER,
+    RELATION_OFFER,
     ClusterSetInstanceState,
     States,
 )
@@ -34,32 +34,34 @@ def setUp(self) -> None:
         self.harness.begin()
         self.peers_relation_id = self.harness.add_relation("database-peers", "db1")
         self.charm = self.harness.charm
-        self.async_primary = self.charm.async_primary
-        self.async_replica = self.charm.async_replica
+        self.async_primary = self.charm.replication_offer
+        self.async_replica = self.charm.replication_consumer
 
     @patch("charm.MySQLOperatorCharm._mysql")
     def test_role(self, _mysql, _):
         _mysql.is_cluster_replica.return_value = True
         _mysql.get_member_state.return_value = (None, "primary")
-        self.async_primary_relation_id = self.harness.add_relation(PRIMARY_RELATION, "db2")
+        self.async_primary_relation_id = self.harness.add_relation(RELATION_OFFER, "db2")
 
         self.assertEqual(
-            self.async_primary.role, ClusterSetInstanceState("replica", "primary", "primary")
+            self.async_primary.role,
+            ClusterSetInstanceState("replica", "primary", "replication-offer"),
         )
         # reset cached value
         del self.async_primary.role
         _mysql.is_cluster_replica.return_value = False
         _mysql.get_member_state.return_value = (None, "secondary")
         self.assertEqual(
-            self.async_primary.role, ClusterSetInstanceState("primary", "secondary", "primary")
+            self.async_primary.role,
+            ClusterSetInstanceState("primary", "secondary", "replication-offer"),
         )
         del self.async_primary.role
 
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_async_relation_broken_primary(self, _mysql, _):
+    def test_async_relation_broken_offer(self, _mysql, _):
         self.harness.set_leader(True)
         self.charm.on.config_changed.emit()
-        self.async_primary_relation_id = self.harness.add_relation(PRIMARY_RELATION, "db2")
+        self.async_primary_relation_id = self.harness.add_relation(RELATION_OFFER, "db2")
         _mysql.is_cluster_replica.return_value = False
         _mysql.get_member_state.return_value = (None, "primary")
         _mysql.is_cluster_in_cluster_set.return_value = True
@@ -78,10 +80,10 @@ def test_async_relation_broken_primary(self, _mysql, _):
         )
 
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_async_relation_broken_replica(self, _mysql, _):
+    def test_async_relation_broken_consumer(self, _mysql, _):
         self.harness.set_leader(True)
         self.charm.on.config_changed.emit()
-        async_primary_relation_id = self.harness.add_relation(PRIMARY_RELATION, "db2")
+        async_primary_relation_id = self.harness.add_relation(RELATION_OFFER, "db2")
         _mysql.is_cluster_replica.return_value = True
         _mysql.get_member_state.return_value = (None, "primary")
         _mysql.is_instance_in_cluster.return_value = False
@@ -95,22 +97,22 @@ def test_async_relation_broken_replica(self, _mysql, _):
 
     @patch("charm.MySQLOperatorCharm._mysql")
     def test_get_state(self, _mysql, _):
-        async_primary_relation_id = self.harness.add_relation(PRIMARY_RELATION, "db2")
-        relation = self.harness.model.get_relation(PRIMARY_RELATION, async_primary_relation_id)
+        async_primary_relation_id = self.harness.add_relation(RELATION_OFFER, "db2")
+        relation = self.harness.model.get_relation(RELATION_OFFER, async_primary_relation_id)
         assert relation
-        self.assertEqual(self.async_primary.get_state(relation), States.UNINITIALIZED)
+        self.assertEqual(self.async_primary.state, States.UNINITIALIZED)
 
         self.harness.update_relation_data(
             async_primary_relation_id, self.charm.app.name, {"is-replica": "true"}
         )
-        self.assertEqual(self.async_primary.get_state(relation), States.FAILED)
+        self.assertEqual(self.async_primary.state, States.FAILED)
 
         self.harness.update_relation_data(
             async_primary_relation_id,
             self.charm.app.name,
             {"is-replica": "", "secret-id": "secret"},
         )
-        self.assertEqual(self.async_primary.get_state(relation), States.SYNCING)
+        self.assertEqual(self.async_primary.state, States.SYNCING)
 
         self.harness.update_relation_data(
             async_primary_relation_id,
@@ -118,44 +120,53 @@ def test_get_state(self, _mysql, _):
             {"endpoint": "db2-endpoint", "cluster-name": "other-cluster"},
         )
         _mysql.get_replica_cluster_status.return_value = "ok"
-        self.assertEqual(self.async_primary.get_state(relation), States.READY)
+        self.assertEqual(self.async_primary.state, States.READY)
 
         _mysql.get_replica_cluster_status.return_value = "unknown"
-        self.assertEqual(self.async_primary.get_state(relation), States.INITIALIZING)
+        self.assertEqual(self.async_primary.state, States.INITIALIZING)
 
         _mysql.get_replica_cluster_status.return_value = "recovering"
-        self.assertEqual(self.async_primary.get_state(relation), States.RECOVERING)
+        self.assertEqual(self.async_primary.state, States.RECOVERING)
 
     @pytest.mark.usefixtures("with_juju_secrets")
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_primary_created(self, _mysql, _):
+    def test_create_replication(self, _mysql, _):
         self.harness.set_leader(True)
         self.charm.on.config_changed.emit()
 
         _mysql.is_cluster_replica.return_value = False
         _mysql.get_mysql_version.return_value = "8.0.36-0ubuntu0.22.04.1"
+        _mysql.get_member_state.return_value = ("online", "primary")
 
         self.harness.update_relation_data(
             self.peers_relation_id, self.charm.unit.name, {"unit-initialized": "True"}
         )
+        self.harness.update_relation_data(
+            self.peers_relation_id, self.charm.app.name, {"units-added-to-cluster": "1"}
+        )
 
         async_primary_relation_id = self.harness.add_relation(
-            PRIMARY_RELATION, "db2", app_data={"is-replica": "true"}
+            RELATION_OFFER, "db2", app_data={"is-replica": "true"}
         )
 
+        self.harness.run_action("create-replication")
+
         relation_data = self.harness.get_relation_data(
             async_primary_relation_id, self.charm.app.name
         )
         self.assertIn("secret-id", relation_data)
         self.assertEqual(relation_data["mysql-version"], "8.0.36-0ubuntu0.22.04.1")
 
-    @patch("charms.mysql.v0.async_replication.MySQLAsyncReplicationPrimary.get_state")
+    @patch(
+        "charms.mysql.v0.async_replication.MySQLAsyncReplicationOffer.state",
+        new_callable=PropertyMock,
+    )
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_primary_relation_changed(self, _mysql, _get_state, _):
+    def test_offer_relation_changed(self, _mysql, _state, _):
         self.harness.set_leader(True)
-        async_primary_relation_id = self.harness.add_relation(PRIMARY_RELATION, "db2")
+        async_primary_relation_id = self.harness.add_relation(RELATION_OFFER, "db2")
 
-        _get_state.return_value = States.INITIALIZING
+        _state.return_value = States.INITIALIZING
 
         # test with donor
         _mysql.get_cluster_endpoints.return_value = (None, "db2-ro-endpoint", None)
@@ -192,7 +203,7 @@ def test_primary_relation_changed(self, _mysql, _get_state, _):
         )
 
         # recovering state
-        _get_state.return_value = States.RECOVERING
+        _state.return_value = States.RECOVERING
         self.harness.update_relation_data(
             async_primary_relation_id,
             "db2",
@@ -208,7 +219,7 @@ def test_state(self, _mysql, _):
         """Test async replica state property."""
         self.assertIsNone(self.async_replica.state)
 
-        async_relation_id = self.harness.add_relation(REPLICA_RELATION, "db1")
+        async_relation_id = self.harness.add_relation(RELATION_CONSUMER, "db1")
 
         # initial state
         self.assertEqual(self.async_replica.state, States.INITIALIZING)
@@ -240,7 +251,7 @@ def test_state(self, _mysql, _):
         self.assertEqual(self.async_replica.state, States.FAILED)
 
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_replica_created(self, _mysql, _):
+    def test_consumer_created(self, _mysql, _):
         """Test replica creation."""
         self.harness.set_leader(True)
         self.charm.on.config_changed.emit()
@@ -251,12 +262,12 @@ def test_replica_created(self, _mysql, _):
 
         _mysql.get_non_system_databases.return_value = set()
 
-        self.harness.add_relation(REPLICA_RELATION, "db1")
+        self.harness.add_relation(RELATION_CONSUMER, "db1")
 
         self.assertTrue(isinstance(self.charm.unit.status, WaitingStatus))
 
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_replica_created_user_data(self, _mysql, _):
+    def test_consumer_created_user_data(self, _mysql, _):
         """Test replica creation."""
         self.harness.set_leader(True)
         self.charm.on.config_changed.emit()
@@ -267,7 +278,7 @@ def test_replica_created_user_data(self, _mysql, _):
 
         _mysql.get_non_system_databases.return_value = set("a-database")
 
-        async_relation_id = self.harness.add_relation(REPLICA_RELATION, "db1")
+        async_relation_id = self.harness.add_relation(RELATION_CONSUMER, "db1")
 
         self.assertIn(
             "user-data-found",
@@ -277,15 +288,15 @@ def test_replica_created_user_data(self, _mysql, _):
     @patch_network_get(private_address="1.1.1.1")
     @patch("ops.framework.EventBase.defer")
     @patch(
-        "charms.mysql.v0.async_replication.MySQLAsyncReplicationReplica.returning_cluster",
+        "charms.mysql.v0.async_replication.MySQLAsyncReplicationConsumer.returning_cluster",
         new_callable=PropertyMock,
     )
     @patch(
-        "charms.mysql.v0.async_replication.MySQLAsyncReplicationReplica.state",
+        "charms.mysql.v0.async_replication.MySQLAsyncReplicationConsumer.state",
         new_callable=PropertyMock,
     )
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_replica_changed_syncing(self, _mysql, _state, _returning_cluster, _defer, _):
+    def test_consumer_changed_syncing(self, _mysql, _state, _returning_cluster, _defer, _):
         """Test replica changed for syncing state."""
         self.harness.set_leader(True)
         self.charm.on.config_changed.emit()
@@ -293,7 +304,7 @@ def test_replica_changed_syncing(self, _mysql, _state, _returning_cluster, _defe
         self.harness.update_relation_data(
             self.peers_relation_id, self.charm.unit.name, {"unit-initialized": "True"}
         )
-        async_relation_id = self.harness.add_relation(REPLICA_RELATION, "db1")
+        async_relation_id = self.harness.add_relation(RELATION_CONSUMER, "db1")
 
         # 1. returning cluster
         _state.return_value = States.SYNCING
@@ -358,18 +369,18 @@ def test_replica_changed_syncing(self, _mysql, _state, _returning_cluster, _defe
 
     @patch("charm.MySQLOperatorCharm._on_update_status")
     @patch(
-        "charms.mysql.v0.async_replication.MySQLAsyncReplicationReplica.state",
+        "charms.mysql.v0.async_replication.MySQLAsyncReplicationConsumer.state",
         new_callable=PropertyMock,
     )
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_replica_changed_ready(self, _mysql, _state, _update_status, _):
+    def test_consumer_changed_ready(self, _mysql, _state, _update_status, _):
         """Test replica changed for ready state."""
         self.harness.set_leader(True)
         self.charm.on.config_changed.emit()
         _state.return_value = States.READY
         _mysql.get_cluster_set_name.return_value = "cluster-set-test"
 
-        async_relation_id = self.harness.add_relation(REPLICA_RELATION, "db1")
+        async_relation_id = self.harness.add_relation(RELATION_CONSUMER, "db1")
         self.harness.update_relation_data(
             async_relation_id,
             "db1",
@@ -382,11 +393,11 @@ def test_replica_changed_ready(self, _mysql, _state, _update_status, _):
 
     @patch("ops.framework.EventBase.defer")
     @patch(
-        "charms.mysql.v0.async_replication.MySQLAsyncReplicationReplica.state",
+        "charms.mysql.v0.async_replication.MySQLAsyncReplicationConsumer.state",
         new_callable=PropertyMock,
     )
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_replica_changed_recovering(self, _mysql, _state, _defer, _):
+    def test_consumer_changed_recovering(self, _mysql, _state, _defer, _):
         """Test replica changed for ready state."""
         self.harness.set_leader(True)
         self.charm.on.config_changed.emit()
@@ -394,7 +405,7 @@ def test_replica_changed_recovering(self, _mysql, _state, _defer, _):
         _mysql.get_cluster_node_count.return_value = 2
 
         with self.harness.hooks_disabled():
-            async_relation_id = self.harness.add_relation(REPLICA_RELATION, "db1")
+            async_relation_id = self.harness.add_relation(RELATION_CONSUMER, "db1")
         self.harness.update_relation_data(
             async_relation_id,
             "db1",
@@ -404,20 +415,20 @@ def test_replica_changed_recovering(self, _mysql, _state, _defer, _):
         self.assertEqual(self.charm.app_peer_data["units-added-to-cluster"], "2")
         _defer.assert_called_once()
 
-    def test_replica_created_non_leader(self, _):
+    def test_consumer_created_non_leader(self, _):
         """Test replica changed for non-leader unit."""
         self.harness.set_leader(False)
         self.charm.unit_peer_data["member-state"] = "online"
-        self.harness.add_relation(REPLICA_RELATION, "db1")
+        self.harness.add_relation(RELATION_CONSUMER, "db1")
 
         self.assertEqual(self.charm.unit_peer_data["member-state"], "waiting")
 
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_replica_changed_non_leader(self, _mysql, _):
+    def test_consumer_changed_non_leader(self, _mysql, _):
         """Test replica changed for non-leader unit."""
         self.harness.set_leader(False)
         with self.harness.hooks_disabled():
-            async_relation_id = self.harness.add_relation(REPLICA_RELATION, "db1")
+            async_relation_id = self.harness.add_relation(RELATION_CONSUMER, "db1")
 
         _mysql.is_instance_in_cluster.return_value = False
 
@@ -431,13 +442,13 @@ def test_replica_changed_non_leader(self, _mysql, _):
 
     # actions
     @patch("charm.MySQLOperatorCharm._mysql")
-    def test_promote_standby_cluster(self, _mysql, _):
+    def test_promote_to_primary(self, _mysql, _):
         self.harness.set_leader(True)
 
         _mysql.is_cluster_replica.return_value = True
 
         self.harness.run_action(
-            "promote-standby-cluster",
+            "promote-to-primary",
             {"cluster-set-name": self.charm.app_peer_data["cluster-set-domain-name"]},
         )
 
@@ -448,7 +459,7 @@ def test_promote_standby_cluster(self, _mysql, _):
         _mysql.reset_mock()
 
         self.harness.run_action(
-            "promote-standby-cluster",
+            "promote-to-primary",
             {
                 "cluster-set-name": self.charm.app_peer_data["cluster-set-domain-name"],
                 "force": True,
@@ -459,70 +470,6 @@ def test_promote_standby_cluster(self, _mysql, _):
             self.charm.app_peer_data["cluster-name"], True
         )
 
-    @patch("charm.MySQLOperatorCharm._on_update_status")
-    @patch("charm.MySQLOperatorCharm._mysql")
-    def test_fence_cluster(self, _mysql, _update_status, _):
-        self.harness.set_leader(True)
-        # fail on wrong cluster set name
-        with self.assertRaises(ActionFailed):
-            self.harness.run_action(
-                "fence-writes",
-                {"cluster-set-name": "incorrect-name"},
-            )
-
-        cluster_set_name = self.charm.app_peer_data["cluster-set-domain-name"]
-        _mysql.get_member_state.return_value = ("online", "primary")
-        _mysql.is_cluster_replica.return_value = True
-        # fail on replica
-        with self.assertRaises(ActionFailed):
-            self.harness.run_action(
-                "fence-writes",
-                {"cluster-set-name": cluster_set_name},
-            )
-
-        del self.async_primary.role
-        _mysql.is_cluster_replica.return_value = False
-        _mysql.is_cluster_writes_fenced.return_value = True
-        # fail on fence already fenced
-        with self.assertRaises(ActionFailed):
-            self.harness.run_action(
-                "fence-writes",
-                {"cluster-set-name": cluster_set_name},
-            )
-
-        _mysql.is_cluster_writes_fenced.return_value = False
-
-        with self.harness.hooks_disabled():
-            self.harness.add_relation(REPLICA_RELATION, "db1")
-
-        self.harness.run_action(
-            "fence-writes",
-            {"cluster-set-name": cluster_set_name},
-        )
-
-        _mysql.fence_writes.assert_called_once()
-        _update_status.assert_called_once()
-
-    @patch("charm.MySQLOperatorCharm._on_update_status")
-    @patch("charm.MySQLOperatorCharm._mysql")
-    def test_unfence_cluster(self, _mysql, _update_status, _):
-        self.harness.set_leader(True)
-
-        _mysql.is_cluster_replica.return_value = False
-        _mysql.get_member_state.return_value = ("online", "primary")
-        _mysql.is_cluster_writes_fenced.return_value = True
-
-        with self.harness.hooks_disabled():
-            self.harness.add_relation(REPLICA_RELATION, "db1")
-
-        self.harness.run_action(
-            "unfence-writes",
-            {"cluster-set-name": self.charm.app_peer_data["cluster-set-domain-name"]},
-        )
-
-        _mysql.unfence_writes.assert_called_once()
-        _update_status.assert_called_once()
-
     @patch("charm.MySQLOperatorCharm._mysql")
     def test_rejoin_cluster_action(self, _mysql, _):
         with self.assertRaises(ActionFailed):
diff --git a/tests/unit/test_mysql.py b/tests/unit/test_mysql.py
index c4bd18a3c..58ffb741a 100644
--- a/tests/unit/test_mysql.py
+++ b/tests/unit/test_mysql.py
@@ -1848,7 +1848,7 @@ def test_create_replica_cluster(self, _run_mysqlsh_script):
             f"repl_cluster = cs.create_replica_cluster('{endpoint}','{replica_cluster_name}', {options})",
             f"repl_cluster.set_instance_option('{endpoint}', 'label', '{instance_label}')",
         )
-        _run_mysqlsh_script.has_calls(
+        _run_mysqlsh_script.assert_has_calls(
             [
                 call("\n".join(commands)),
                 call("\n".join(commands2)),