diff --git a/actions.yaml b/actions.yaml
index e7ad22d388..70b296e77a 100644
--- a/actions.yaml
+++ b/actions.yaml
@@ -45,6 +45,9 @@ restore:
     backup-id:
       type: string
       description: A backup-id to identify the backup to restore (format = %Y-%m-%dT%H:%M:%SZ)
+    restore-to-time:
+      type: string
+      description: Point-in-time-recovery target in PSQL format.
 resume-upgrade:
   description: Resume a rolling upgrade after asserting successful upgrade of a new revision.
 set-password:
diff --git a/src/backups.py b/src/backups.py
index 6a071b12dc..245e97d864 100644
--- a/src/backups.py
+++ b/src/backups.py
@@ -35,11 +35,14 @@
     "failed to access/create the bucket, check your S3 settings"
 )
 FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE = "failed to initialize stanza, check your S3 settings"
+CANNOT_RESTORE_PITR = "cannot restore PITR, juju debug-log for details"
+MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET = "Move restored cluster to another S3 bucket"
 
 S3_BLOCK_MESSAGES = [
     ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE,
     FAILED_TO_ACCESS_CREATE_BUCKET_ERROR_MESSAGE,
     FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE,
+    MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
 ]
 
 
@@ -168,9 +171,29 @@ def can_use_s3_repository(self) -> Tuple[bool, Optional[str]]:
                     if self.charm._patroni.member_started:
                         self.charm._patroni.reload_patroni_configuration()
                     return False, ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE
+                return self._is_s3_wal_compatible(stanza)
 
         return True, None
 
+    def _is_s3_wal_compatible(self, stanza) -> Tuple[bool, Optional[str]]:
+        """Returns whether the S3 stanza is compatible with current PostgreSQL cluster by WAL parity."""
+        charm_last_archived_wal = self.charm.postgresql.get_last_archived_wal()
+        logger.debug(f"last archived wal: {charm_last_archived_wal}")
+        s3_archive = stanza.get("archive", [])
+        if len(s3_archive) > 0:
+            s3_last_archived_wal = s3_archive[0].get("max")
+            logger.debug(f"last s3 wal: {str(s3_last_archived_wal)}")
+            if (
+                charm_last_archived_wal
+                and s3_last_archived_wal
+                and charm_last_archived_wal.split(".", 1)[0] != str(s3_last_archived_wal)
+            ):
+                if bool(self.charm.app_peer_data.get("require-change-bucket-after-restore", None)):
+                    return False, MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET
+                else:
+                    return False, ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE
+        return True, None
+
     def _construct_endpoint(self, s3_parameters: Dict) -> str:
         """Construct the S3 service endpoint using the region.
 
@@ -240,10 +263,12 @@ def _empty_data_files(self) -> None:
         try:
             self.container.exec("rm -r /var/lib/postgresql/data/pgdata".split()).wait_output()
         except ExecError as e:
-            logger.exception(
-                "Failed to empty data directory in prep for backup restore", exc_info=e
-            )
-            raise
+            # If previous PITR restore was unsuccessful, there is no such directory.
+            if "No such file or directory" not in e.stderr:
+                logger.exception(
+                    "Failed to empty data directory in prep for backup restore", exc_info=e
+                )
+                raise
 
     def _change_connectivity_to_database(self, connectivity: bool) -> None:
         """Enable or disable the connectivity to the database."""
@@ -419,11 +444,7 @@ def _initialise_stanza(self) -> None:
 
         # Enable stanza initialisation if the backup settings were fixed after being invalid
         # or pointing to a repository where there are backups from another cluster.
-        if self.charm.is_blocked and self.charm.unit.status.message not in [
-            ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE,
-            FAILED_TO_ACCESS_CREATE_BUCKET_ERROR_MESSAGE,
-            FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE,
-        ]:
+        if self.charm.is_blocked and self.charm.unit.status.message not in S3_BLOCK_MESSAGES:
             logger.warning("couldn't initialize stanza due to a blocked status")
             return
 
@@ -545,6 +566,21 @@ def _on_s3_credential_changed(self, event: CredentialsChangedEvent):
             event.defer()
             return
 
+        # Prevents config change in bad state, so DB peer relations change event will not cause patroni related errors.
+        if self.charm.unit.status.message == CANNOT_RESTORE_PITR:
+            logger.info("Cannot change S3 configuration in bad PITR restore status")
+            event.defer()
+            return
+
+        # Prevents S3 change in the middle of restoring backup and patroni / pgbackrest errors caused by that.
+        if (
+            "restoring-backup" in self.charm.app_peer_data
+            or "restore-to-time" in self.charm.app_peer_data
+        ):
+            logger.info("Cannot change S3 configuration during restore")
+            event.defer()
+            return
+
         if not self._render_pgbackrest_conf_file():
             logger.debug("Cannot set pgBackRest configurations, missing configurations.")
             return
@@ -554,6 +590,9 @@ def _on_s3_credential_changed(self, event: CredentialsChangedEvent):
             event.defer()
             return
 
+        if self.charm.unit.is_leader():
+            self.charm.app_peer_data.pop("require-change-bucket-after-restore", None)
+
         # Verify the s3 relation only on the primary.
         if not self.charm.is_primary:
             return
@@ -712,7 +751,11 @@ def _on_create_backup_action(self, event) -> None:  # noqa: C901
 
     def _on_s3_credential_gone(self, _) -> None:
         if self.charm.unit.is_leader():
-            self.charm.app_peer_data.update({"stanza": "", "init-pgbackrest": ""})
+            self.charm.app_peer_data.update({
+                "stanza": "",
+                "init-pgbackrest": "",
+                "require-change-bucket-after-restore": "",
+            })
         self.charm.unit_peer_data.update({"stanza": "", "init-pgbackrest": ""})
         if self.charm.is_blocked and self.charm.unit.status.message in S3_BLOCK_MESSAGES:
             self.charm.unit.status = ActiveStatus()
@@ -738,19 +781,52 @@ def _on_restore_action(self, event):
             return
 
         backup_id = event.params.get("backup-id")
-        logger.info(f"A restore with backup-id {backup_id} has been requested on unit")
+        restore_to_time = event.params.get("restore-to-time")
+        logger.info(
+            f"A restore"
+            f"{' with backup-id ' + backup_id if backup_id else ''}"
+            f"{' to time point ' + restore_to_time if restore_to_time else ''}"
+            f" has been requested on the unit"
+        )
 
-        # Validate the provided backup id.
-        logger.info("Validating provided backup-id")
+        # Validate the provided backup id and restore to time.
+        logger.info("Validating provided backup-id and restore-to-time")
         backups = self._list_backups(show_failed=False)
-        if backup_id not in backups.keys():
+        if backup_id and backup_id not in backups.keys():
             error_message = f"Invalid backup-id: {backup_id}"
             logger.error(f"Restore failed: {error_message}")
             event.fail(error_message)
             return
+        if not backup_id and restore_to_time and not backups:
+            error_message = "Cannot restore PITR without any backups created"
+            logger.error(f"Restore failed: {error_message}")
+            event.fail(error_message)
+            return
+
+        # Quick check for timestamp format
+        if (
+            restore_to_time
+            and restore_to_time != "latest"
+            and not re.match("^[0-9-]+ [0-9:.+]+$", restore_to_time)
+        ):
+            error_message = "Bad restore-to-time format"
+            logger.error(f"Restore failed: {error_message}")
+            event.fail(error_message)
+            return
 
         self.charm.unit.status = MaintenanceStatus("restoring backup")
 
+        # Temporarily disabling patroni (postgresql) pebble service auto-restart on failures. This is required
+        # as point-in-time-recovery can fail on restore, therefore during cluster bootstrapping process. In this
+        # case, we need be able to check patroni service status and logs. Disabling auto-restart feature is essential
+        # to prevent wrong status indicated and logs reading race condition (as logs cleared / moved with service
+        # restarts).
+        if not self.charm.override_patroni_on_failure_condition("ignore", "restore-backup"):
+            error_message = "Failed to override Patroni on-failure condition"
+            logger.error(f"Restore failed: {error_message}")
+            event.fail(error_message)
+            return
+
         # Stop the database service before performing the restore.
         logger.info("Stopping database service")
         try:
@@ -778,11 +854,15 @@ def _on_restore_action(self, event):
                 namespace=self.charm._namespace,
             )
         except ApiError as e:
-            error_message = f"Failed to remove previous cluster information with error: {str(e)}"
-            logger.error(f"Restore failed: {error_message}")
-            event.fail(error_message)
-            self._restart_database()
-            return
+            # If previous PITR restore was unsuccessful, there are no such endpoints.
+            if "restore-to-time" not in self.charm.app_peer_data:
+                error_message = (
+                    f"Failed to remove previous cluster information with error: {str(e)}"
+                )
+                logger.error(f"Restore failed: {error_message}")
+                event.fail(error_message)
+                self._restart_database()
+                return
 
         logger.info("Removing the contents of the data directory")
         try:
@@ -800,8 +880,12 @@ def _on_restore_action(self, event):
         # Mark the cluster as in a restoring backup state and update the Patroni configuration.
         logger.info("Configuring Patroni to restore the backup")
         self.charm.app_peer_data.update({
-            "restoring-backup": self._fetch_backup_from_id(backup_id),
-            "restore-stanza": backups[backup_id],
+            "restoring-backup": self._fetch_backup_from_id(backup_id) if backup_id else "",
+            "restore-stanza": backups[backup_id]
+            if backup_id
+            else self.charm.app_peer_data.get("stanza", self.stanza_name),
+            "restore-to-time": restore_to_time or "",
+            "require-change-bucket-after-restore": "True",
         })
         self.charm.update_config()
 
@@ -854,8 +938,10 @@ def _pre_restore_checks(self, event: ActionEvent) -> bool:
             event.fail(validation_message)
             return False
 
-        if not event.params.get("backup-id"):
-            error_message = "Missing backup-id to restore"
+        if not event.params.get("backup-id") and not event.params.get("restore-to-time"):
+            error_message = (
+                "Missing backup-id or/and restore-to-time parameter to be able to do restore"
+            )
             logger.error(f"Restore failed: {error_message}")
             event.fail(error_message)
             return False
@@ -867,10 +953,11 @@ def _pre_restore_checks(self, event: ActionEvent) -> bool:
             return False
 
         logger.info("Checking if cluster is in blocked state")
-        if (
-            self.charm.is_blocked
-            and self.charm.unit.status.message != ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE
-        ):
+        if self.charm.is_blocked and self.charm.unit.status.message not in [
+            ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE,
+            CANNOT_RESTORE_PITR,
+            MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
+        ]:
             error_message = "Cluster or unit is in a blocking state"
             logger.error(f"Restore failed: {error_message}")
             event.fail(error_message)
@@ -956,7 +1043,7 @@ def _render_pgbackrest_conf_file(self) -> bool:
 
     def _restart_database(self) -> None:
         """Removes the restoring backup flag and restart the database."""
-        self.charm.app_peer_data.update({"restoring-backup": ""})
+        self.charm.app_peer_data.update({"restoring-backup": "", "restore-to-time": ""})
         self.charm.update_config()
         self.container.start(self.charm._postgresql_service)
 
diff --git a/src/charm.py b/src/charm.py
index 4014617ffc..6d0fdaf15e 100755
--- a/src/charm.py
+++ b/src/charm.py
@@ -8,6 +8,7 @@
 import json
 import logging
 import os
+import re
 import sys
 from pathlib import Path
 from typing import Dict, List, Literal, Optional, Tuple, get_args
@@ -53,11 +54,11 @@
     UnknownStatus,
     WaitingStatus,
 )
-from ops.pebble import ChangeError, Layer, PathError, ProtocolError, ServiceStatus
+from ops.pebble import ChangeError, Layer, PathError, ProtocolError, ServiceInfo, ServiceStatus
 from requests import ConnectionError
 from tenacity import RetryError, Retrying, stop_after_attempt, stop_after_delay, wait_fixed
 
-from backups import PostgreSQLBackups
+from backups import CANNOT_RESTORE_PITR, MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET, PostgreSQLBackups
 from config import CharmConfig
 from constants import (
     APP_SCOPE,
@@ -101,6 +102,8 @@
 EXTENSIONS_DEPENDENCY_MESSAGE = "Unsatisfied plugin dependencies. Please check the logs"
 EXTENSION_OBJECT_MESSAGE = "Cannot disable plugins: Existing objects depend on it. See logs"
 
+ORIGINAL_PATRONI_ON_FAILURE_CONDITION = "restart"
+
 # http{x,core} clutter the logs with debug messages
 logging.getLogger("httpcore").setLevel(logging.ERROR)
 logging.getLogger("httpx").setLevel(logging.ERROR)
@@ -505,6 +508,13 @@ def _on_peer_relation_changed(self, event: HookEvent) -> None:  # noqa: C901
             logger.error("Invalid configuration: %s", str(e))
             return
 
+        # If PITR restore failed, then wait it for resolve.
+        if (
+            "restoring-backup" in self.app_peer_data or "restore-to-time" in self.app_peer_data
+        ) and isinstance(self.unit.status, BlockedStatus):
+            event.defer()
+            return
+
         # Validate the status of the member before setting an ActiveStatus.
         if not self._patroni.member_started:
             logger.debug("Deferring on_peer_relation_changed: Waiting for member to start")
@@ -897,6 +907,15 @@ def _on_postgresql_pebble_ready(self, event: WorkloadEvent) -> None:
         self._set_active_status()
 
     def _set_active_status(self):
+        if "require-change-bucket-after-restore" in self.app_peer_data:
+            if self.unit.is_leader():
+                self.app_peer_data.update({
+                    "restoring-backup": "",
+                    "restore-stanza": "",
+                    "restore-to-time": "",
+                })
+            self.unit.status = BlockedStatus(MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET)
+            return
         try:
             if self._patroni.get_primary(unit_name_pattern=True) == self.unit.name:
                 self.unit.status = ActiveStatus("Primary")
@@ -1090,9 +1109,13 @@ def _has_blocked_status(self) -> bool:
         return isinstance(self.unit.status, BlockedStatus)
 
     @property
-    def _has_waiting_status(self) -> bool:
-        """Returns whether the unit is in a waiting state."""
-        return isinstance(self.unit.status, WaitingStatus)
+    def _has_non_restore_waiting_status(self) -> bool:
+        """Returns whether the unit is in a waiting state and there is no restore process ongoing."""
+        return (
+            isinstance(self.unit.status, WaitingStatus)
+            and "restoring-backup" not in self.app_peer_data
+            and "restore-to-time" not in self.app_peer_data
+        )
 
     def _on_get_password(self, event: ActionEvent) -> None:
         """Returns the password for a user as an action response.
@@ -1273,7 +1296,7 @@ def _on_update_status(self, _) -> None:
             logger.debug("on_update_status early exit: Cannot connect to container")
             return
 
-        if self._has_blocked_status or self._has_waiting_status:
+        if self._has_blocked_status or self._has_non_restore_waiting_status:
             # If charm was failing to disable plugin, try again (user may have removed the objects)
             if self.unit.status.message == EXTENSION_OBJECT_MESSAGE:
                 self.enable_disable_extensions()
@@ -1300,9 +1323,9 @@ def _on_update_status(self, _) -> None:
                 self.unit.status = MaintenanceStatus("Database service inactive, restarting")
                 return
 
-        if "restoring-backup" in self.app_peer_data and not self._was_restore_successful(
-            services[0]
-        ):
+        if (
+            "restoring-backup" in self.app_peer_data or "restore-to-time" in self.app_peer_data
+        ) and not self._was_restore_successful(container, services[0]):
             return
 
         if self._handle_processes_failures():
@@ -1313,9 +1336,18 @@ def _on_update_status(self, _) -> None:
 
         self._set_active_status()
 
-    def _was_restore_successful(self, service) -> bool:
+    def _was_restore_successful(self, container: Container, service: ServiceInfo) -> bool:
         """Checks if restore operation succeeded and S3 is properly configured."""
         if service.current != ServiceStatus.ACTIVE:
+            if "restore-to-time" in self.app_peer_data and all(self.is_pitr_failed(container)):
+                logger.error(
+                    "Restore failed: database service failed to reach point-in-time-recovery target. "
+                    "You can launch another restore with different parameters"
+                )
+                self.log_pitr_last_transaction_time()
+                self.unit.status = BlockedStatus(CANNOT_RESTORE_PITR)
+                return False
+
             logger.error("Restore failed: database service failed to start")
             self.unit.status = BlockedStatus("Failed to restore backup")
             return False
@@ -1325,8 +1357,13 @@ def _was_restore_successful(self, service) -> bool:
             return False
 
         # Remove the restoring backup flag and the restore stanza name.
-        self.app_peer_data.update({"restoring-backup": "", "restore-stanza": ""})
+        self.app_peer_data.update({
+            "restoring-backup": "",
+            "restore-stanza": "",
+            "restore-to-time": "",
+        })
         self.update_config()
+        self.restore_patroni_on_failure_condition()
         logger.info("Restore succeeded")
 
         can_use_s3_repository, validation_message = self.backup.can_use_s3_repository()
@@ -1503,6 +1540,10 @@ def _postgresql_layer(self) -> Layer:
                     "summary": "entrypoint of the postgresql + patroni image",
                     "command": f"patroni {self._storage_path}/patroni.yml",
                     "startup": "enabled",
+                    "on-failure": self.unit_peer_data.get(
+                        "patroni-on-failure-condition-override", None
+                    )
+                    or ORIGINAL_PATRONI_ON_FAILURE_CONDITION,
                     "user": WORKLOAD_OS_USER,
                     "group": WORKLOAD_OS_GROUP,
                     "environment": {
@@ -1682,8 +1723,13 @@ def update_config(self, is_creating_backup: bool = False) -> bool:
             enable_tls=self.is_tls_enabled,
             is_no_sync_member=self.upgrade.is_no_sync_member,
             backup_id=self.app_peer_data.get("restoring-backup"),
+            pitr_target=self.app_peer_data.get("restore-to-time"),
+            restore_to_latest=self.app_peer_data.get("restore-to-time", None) == "latest",
             stanza=self.app_peer_data.get("stanza"),
             restore_stanza=self.app_peer_data.get("restore-stanza"),
+            disable_pgbackrest_archiving=bool(
+                self.app_peer_data.get("require-change-bucket-after-restore", None)
+            ),
             parameters=postgresql_parameters,
         )
 
@@ -1783,7 +1829,7 @@ def _handle_postgresql_restart_need(self):
             )
             self.on[self.restart_manager.name].acquire_lock.emit()
 
-    def _update_pebble_layers(self) -> None:
+    def _update_pebble_layers(self, replan: bool = True) -> None:
         """Update the pebble layers to keep the health check URL up-to-date."""
         container = self.unit.get_container("postgresql")
 
@@ -1798,8 +1844,9 @@ def _update_pebble_layers(self) -> None:
             # Changes were made, add the new layer.
             container.add_layer(self._postgresql_service, new_layer, combine=True)
             logging.info("Added updated layer 'postgresql' to Pebble plan")
-            container.replan()
-            logging.info("Restarted postgresql service")
+            if replan:
+                container.replan()
+                logging.info("Restarted postgresql service")
         if current_layer.checks != new_layer.checks:
             # Changes were made, add the new layer.
             container.add_layer(self._postgresql_service, new_layer, combine=True)
@@ -1893,6 +1940,113 @@ def client_relations(self) -> List[Relation]:
                 relations.append(relation)
         return relations
 
+    def override_patroni_on_failure_condition(
+        self, new_condition: str, repeat_cause: str | None
+    ) -> bool:
+        """Temporary override Patroni pebble service on-failure condition.
+
+        Executes only on current unit.
+
+        Args:
+            new_condition: new Patroni pebble service on-failure condition.
+            repeat_cause: whether this field is equal to the last success override operation repeat cause, Patroni
+                on-failure condition will be overridden (keeping the original restart condition reference untouched) and
+                success code will be returned. But if this field is distinct from previous repeat cause or None,
+                repeated operation will cause failure code will be returned.
+        """
+        if "patroni-on-failure-condition-override" in self.unit_peer_data:
+            current_condition = self.unit_peer_data["patroni-on-failure-condition-override"]
+            if repeat_cause is None:
+                logger.error(
+                    f"failure trying to override patroni on-failure condition to {new_condition}"
+                    f"as it already overridden from {ORIGINAL_PATRONI_ON_FAILURE_CONDITION} to {current_condition}"
+                )
+                return False
+            previous_repeat_cause = self.unit_peer_data.get(
+                "overridden-patroni-on-failure-condition-repeat-cause", None
+            )
+            if previous_repeat_cause != repeat_cause:
+                logger.error(
+                    f"failure trying to override patroni on-failure condition to {new_condition}"
+                    f"as it already overridden from {ORIGINAL_PATRONI_ON_FAILURE_CONDITION} to {current_condition}"
+                    f"and repeat cause is not equal: {previous_repeat_cause} != {repeat_cause}"
+                )
+                return False
+            self.unit_peer_data["patroni-on-failure-condition-override"] = new_condition
+            self._update_pebble_layers(False)
+            logger.debug(
+                f"Patroni on-failure condition re-overridden to {new_condition} within repeat cause {repeat_cause}"
+                f"(original on-failure condition reference is untouched and is {ORIGINAL_PATRONI_ON_FAILURE_CONDITION})"
+            )
+            return True
+
+        self.unit_peer_data["patroni-on-failure-condition-override"] = new_condition
+        if repeat_cause:
+            self.unit_peer_data["overridden-patroni-on-failure-condition-repeat-cause"] = (
+                repeat_cause
+            )
+        self._update_pebble_layers(False)
+        logger.debug(
+            f"Patroni on-failure condition overridden from {ORIGINAL_PATRONI_ON_FAILURE_CONDITION} to {new_condition}"
+            f"{' with repeat cause ' + repeat_cause if repeat_cause is not None else ''}"
+        )
+        return True
+
+    def restore_patroni_on_failure_condition(self) -> None:
+        """Restore Patroni pebble service original on-failure condition.
+
+        Will do nothing if not overridden. Executes only on current unit.
+        """
+        if "patroni-on-failure-condition-override" in self.unit_peer_data:
+            self.unit_peer_data.update({
+                "patroni-on-failure-condition-override": "",
+                "overridden-patroni-on-failure-condition-repeat-cause": "",
+            })
+            self._update_pebble_layers(False)
+            logger.debug(
+                f"restored Patroni on-failure condition to {ORIGINAL_PATRONI_ON_FAILURE_CONDITION}"
+            )
+        else:
+            logger.warning("not restoring patroni on-failure condition as it's not overridden")
+
+    def is_pitr_failed(self, container: Container) -> Tuple[bool, bool]:
+        """Check if Patroni service failed to bootstrap cluster during point-in-time-recovery.
+
+        Typically, this means that database service failed to reach point-in-time-recovery target or has been
+        supplied with bad PITR parameter. Also, remembers last state and can provide info is it new event, or
+        it belongs to previous action. Executes only on current unit.
+
+        Returns:
+            Tuple[bool, bool]:
+                - Is patroni service failed to bootstrap cluster.
+                - Is it new fail, that wasn't observed previously.
+        """
+        log_exec = container.pebble.exec(["pebble", "logs", "postgresql"], combine_stderr=True)
+        patroni_logs = log_exec.wait_output()[0]
+        patroni_exceptions = re.findall(
+            r"^([0-9-:TZ.]+) \[postgresql] patroni\.exceptions\.PatroniFatalException: Failed to bootstrap cluster$",
+            patroni_logs,
+            re.MULTILINE,
+        )
+        if len(patroni_exceptions) > 0:
+            old_pitr_fail_id = self.unit_peer_data.get("last_pitr_fail_id", None)
+            self.unit_peer_data["last_pitr_fail_id"] = patroni_exceptions[-1]
+            return True, patroni_exceptions[-1] != old_pitr_fail_id
+        return False, False
+
+    def log_pitr_last_transaction_time(self) -> None:
+        """Log to user last completed transaction time acquired from postgresql logs."""
+        postgresql_logs = self._patroni.last_postgresql_logs()
+        log_time = re.findall(
+            r"last completed transaction was at log time (.*)$",
+            postgresql_logs,
+            re.MULTILINE,
+        )
+        if len(log_time) > 0:
+            logger.info(f"Last completed transaction was at {log_time[-1]}")
+        else:
+            logger.error("Can't tell last completed transaction time")
+
 
 if __name__ == "__main__":
     main(PostgresqlOperatorCharm, use_juju_for_storage=True)
diff --git a/src/constants.py b/src/constants.py
index 5f1927b298..8dc02738bd 100644
--- a/src/constants.py
+++ b/src/constants.py
@@ -21,6 +21,8 @@
 WORKLOAD_OS_USER = "postgres"
 METRICS_PORT = "9187"
 POSTGRESQL_DATA_PATH = "/var/lib/postgresql/data/pgdata"
+POSTGRESQL_LOGS_PATH = "/var/log/postgresql"
+POSTGRESQL_LOGS_PATTERN = "postgresql*.log"
 POSTGRES_LOG_FILES = [
     "/var/log/pgbackrest/*",
     "/var/log/postgresql/patroni.log",
diff --git a/src/patroni.py b/src/patroni.py
index 93e1fd1948..91a1a4af72 100644
--- a/src/patroni.py
+++ b/src/patroni.py
@@ -24,7 +24,7 @@
     wait_fixed,
 )
 
-from constants import REWIND_USER, TLS_CA_FILE
+from constants import POSTGRESQL_LOGS_PATH, POSTGRESQL_LOGS_PATTERN, REWIND_USER, TLS_CA_FILE
 
 RUNNING_STATES = ["running", "streaming"]
 
@@ -404,7 +404,10 @@ def render_patroni_yml_file(
         is_no_sync_member: bool = False,
         stanza: str = None,
         restore_stanza: Optional[str] = None,
+        disable_pgbackrest_archiving: bool = False,
         backup_id: Optional[str] = None,
+        pitr_target: Optional[str] = None,
+        restore_to_latest: bool = False,
         parameters: Optional[dict[str, str]] = None,
     ) -> None:
         """Render the Patroni configuration file.
@@ -417,7 +420,10 @@ def render_patroni_yml_file(
                 (when it's a replica).
             stanza: name of the stanza created by pgBackRest.
             restore_stanza: name of the stanza used when restoring a backup.
+            disable_pgbackrest_archiving: whether to force disable pgBackRest WAL archiving.
             backup_id: id of the backup that is being restored.
+            pitr_target: point-in-time-recovery target for the backup.
+            restore_to_latest: restore all the WAL transaction logs from the stanza.
             parameters: PostgreSQL parameters to be added to the postgresql.conf file.
         """
         # Open the template patroni.yml file.
@@ -437,9 +443,12 @@ def render_patroni_yml_file(
             replication_password=self._replication_password,
             rewind_user=REWIND_USER,
             rewind_password=self._rewind_password,
-            enable_pgbackrest=stanza is not None,
-            restoring_backup=backup_id is not None,
+            enable_pgbackrest_archiving=stanza is not None
+            and disable_pgbackrest_archiving is False,
+            restoring_backup=backup_id is not None or pitr_target is not None,
             backup_id=backup_id,
+            pitr_target=pitr_target if not restore_to_latest else None,
+            restore_to_latest=restore_to_latest,
             stanza=stanza,
             restore_stanza=restore_stanza,
             minority_count=self._members_count // 2,
@@ -455,6 +464,30 @@ def reload_patroni_configuration(self) -> None:
         """Reloads the configuration after it was updated in the file."""
         requests.post(f"{self._patroni_url}/reload", verify=self._verify)
 
+    def last_postgresql_logs(self) -> str:
+        """Get last log file content of Postgresql service in the container.
+
+        If there is no available log files, empty line will be returned.
+
+        Returns:
+            Content of last log file of Postgresql service.
+        """
+        container = self._charm.unit.get_container("postgresql")
+        if not container.can_connect():
+            logger.debug("Cannot get last PostgreSQL log from Rock. Container inaccessible")
+            return ""
+        log_files = container.list_files(POSTGRESQL_LOGS_PATH, pattern=POSTGRESQL_LOGS_PATTERN)
+        if len(log_files) == 0:
+            return ""
+        log_files.sort(key=lambda f: f.path, reverse=True)
+        try:
+            with container.pull(log_files[0].path) as last_log_file:
+                return last_log_file.read()
+        except OSError as e:
+            error_message = "Failed to read last postgresql log file"
+            logger.exception(error_message, exc_info=e)
+            return ""
+
     @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
     def restart_postgresql(self) -> None:
         """Restart PostgreSQL."""
diff --git a/templates/patroni.yml.j2 b/templates/patroni.yml.j2
index 0ef6ec2c5e..fbefb2294c 100644
--- a/templates/patroni.yml.j2
+++ b/templates/patroni.yml.j2
@@ -9,7 +9,7 @@ bootstrap:
       bin_dir: /usr/lib/postgresql/{{ version }}/bin
       parameters:
         synchronous_standby_names: "*"
-        {%- if enable_pgbackrest %}
+        {%- if enable_pgbackrest_archiving %}
         archive_command: 'pgbackrest --stanza={{ stanza }} archive-push %p'
         {% else %}
         archive_command: /bin/true
@@ -49,7 +49,13 @@ bootstrap:
   {%- if restoring_backup %}
   method: pgbackrest
   pgbackrest:
-    command: pgbackrest --stanza={{ restore_stanza }} --pg1-path={{ storage_path }}/pgdata --set={{ backup_id }} --type=immediate --target-action=promote restore
+    command: >
+      pgbackrest --stanza={{ restore_stanza }} --pg1-path={{ storage_path }}/pgdata
+      {%- if backup_id %} --set={{ backup_id }} {%- endif %}
+      {%- if restore_to_latest %} --type=default {%- else %}
+      --target-action=promote {%- if pitr_target %} --target="{{ pitr_target }}" --type=time {%- else %} --type=immediate {%- endif %}
+      {%- endif %}
+      restore
     no_params: True
     keep_existing_recovery_conf: True
   {% elif primary_cluster_endpoint %}
@@ -98,7 +104,7 @@ postgresql:
   bin_dir: /usr/lib/postgresql/{{ version }}/bin
   listen: 0.0.0.0:5432
   parameters:
-    {%- if enable_pgbackrest %}
+    {%- if enable_pgbackrest_archiving %}
     archive_command: 'pgbackrest --stanza={{ stanza }} archive-push %p'
     {% else %}
     archive_command: /bin/true
diff --git a/tests/integration/helpers.py b/tests/integration/helpers.py
index 3b1a61b72e..1c8f1bc6aa 100644
--- a/tests/integration/helpers.py
+++ b/tests/integration/helpers.py
@@ -37,6 +37,7 @@
 DATABASE_APP_NAME = METADATA["name"]
 APPLICATION_NAME = "postgresql-test-app"
 STORAGE_PATH = METADATA["storage"]["pgdata"]["location"]
+MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET = "Move restored cluster to another S3 bucket"
 
 charm = None
 
@@ -920,6 +921,8 @@ async def backup_operations(
     async with ops_test.fast_forward(fast_interval="60s"):
         await scale_application(ops_test, database_app_name, 1)
 
+    remaining_unit = ops_test.model.units.get(f"{database_app_name}/0")
+
     # Run the "restore backup" action for differential backup.
     for attempt in Retrying(
         stop=stop_after_attempt(10), wait=wait_exponential(multiplier=1, min=2, max=30)
@@ -928,16 +931,18 @@ async def backup_operations(
             logger.info("restoring the backup")
             last_diff_backup = backups.split("\n")[-1]
             backup_id = last_diff_backup.split()[0]
-            action = await ops_test.model.units.get(f"{database_app_name}/0").run_action(
-                "restore", **{"backup-id": backup_id}
-            )
+            action = await remaining_unit.run_action("restore", **{"backup-id": backup_id})
             await action.wait()
             restore_status = action.results.get("restore-status")
             assert restore_status, "restore hasn't succeeded"
 
     # Wait for the restore to complete.
     async with ops_test.fast_forward():
-        await ops_test.model.wait_for_idle(status="active", timeout=1000)
+        await ops_test.model.block_until(
+            lambda: remaining_unit.workload_status_message
+            == MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
+            timeout=1000,
+        )
 
     # Check that the backup was correctly restored by having only the first created table.
     logger.info("checking that the backup was correctly restored")
@@ -975,16 +980,18 @@ async def backup_operations(
             logger.info("restoring the backup")
             last_full_backup = backups.split("\n")[-2]
             backup_id = last_full_backup.split()[0]
-            action = await ops_test.model.units.get(f"{database_app_name}/0").run_action(
-                "restore", **{"backup-id": backup_id}
-            )
+            action = await remaining_unit.run_action("restore", **{"backup-id": backup_id})
             await action.wait()
             restore_status = action.results.get("restore-status")
             assert restore_status, "restore hasn't succeeded"
 
     # Wait for the restore to complete.
     async with ops_test.fast_forward():
-        await ops_test.model.wait_for_idle(status="active", timeout=1000)
+        await ops_test.model.block_until(
+            lambda: remaining_unit.workload_status_message
+            == MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
+            timeout=1000,
+        )
 
     # Check that the backup was correctly restored by having only the first created table.
     logger.info("checking that the backup was correctly restored")
diff --git a/tests/integration/test_backups.py b/tests/integration/test_backups.py
index dd8c355fa6..f4425391b9 100644
--- a/tests/integration/test_backups.py
+++ b/tests/integration/test_backups.py
@@ -15,6 +15,7 @@
 from . import architecture
 from .helpers import (
     DATABASE_APP_NAME,
+    MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
     backup_operations,
     build_and_deploy,
     cat_file_from_unit,
@@ -23,7 +24,6 @@
     get_password,
     get_primary,
     get_unit_address,
-    scale_application,
     switchover,
     wait_for_idle_on_blocked,
 )
@@ -126,34 +126,43 @@ async def test_backup_aws(ops_test: OpsTest, cloud_configs: Tuple[Dict, Dict]) -
             f"{database_app_name}:certificates",
             f"{tls_certificates_app_name}:certificates",
         )
-        await ops_test.model.wait_for_idle(apps=[database_app_name], status="active", timeout=1000)
+
+        new_unit_name = f"{database_app_name}/1"
 
         # Scale up to be able to test primary and leader being different.
-        async with ops_test.fast_forward(fast_interval="60s"):
-            await scale_application(ops_test, database_app_name, 2)
+        await ops_test.model.applications[database_app_name].scale(2)
+        await ops_test.model.block_until(
+            lambda: len(ops_test.model.applications[database_app_name].units) == 2
+            and ops_test.model.units.get(new_unit_name).workload_status_message
+            == MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
+            timeout=1000,
+        )
 
         logger.info("ensuring that the replication is working correctly")
-        new_unit_name = f"{database_app_name}/1"
         address = await get_unit_address(ops_test, new_unit_name)
         password = await get_password(ops_test, database_app_name=database_app_name)
-        with db_connect(
-            host=address, password=password
-        ) as connection, connection.cursor() as cursor:
-            cursor.execute(
-                "SELECT EXISTS (SELECT FROM information_schema.tables"
-                " WHERE table_schema = 'public' AND table_name = 'backup_table_1');"
-            )
-            assert cursor.fetchone()[
-                0
-            ], f"replication isn't working correctly: table 'backup_table_1' doesn't exist in {new_unit_name}"
-            cursor.execute(
-                "SELECT EXISTS (SELECT FROM information_schema.tables"
-                " WHERE table_schema = 'public' AND table_name = 'backup_table_2');"
-            )
-            assert not cursor.fetchone()[
-                0
-            ], f"replication isn't working correctly: table 'backup_table_2' exists in {new_unit_name}"
-        connection.close()
+        for attempt in Retrying(
+            stop=stop_after_attempt(10), wait=wait_exponential(multiplier=1, min=2, max=30)
+        ):
+            with attempt:
+                with db_connect(
+                    host=address, password=password
+                ) as connection, connection.cursor() as cursor:
+                    cursor.execute(
+                        "SELECT EXISTS (SELECT FROM information_schema.tables"
+                        " WHERE table_schema = 'public' AND table_name = 'backup_table_1');"
+                    )
+                    assert cursor.fetchone()[
+                        0
+                    ], f"replication isn't working correctly: table 'backup_table_1' doesn't exist in {new_unit_name}"
+                    cursor.execute(
+                        "SELECT EXISTS (SELECT FROM information_schema.tables"
+                        " WHERE table_schema = 'public' AND table_name = 'backup_table_2');"
+                    )
+                    assert not cursor.fetchone()[
+                        0
+                    ], f"replication isn't working correctly: table 'backup_table_2' exists in {new_unit_name}"
+                connection.close()
 
         old_primary = await get_primary(ops_test, database_app_name)
         logger.info(f"performing a switchover from {old_primary} to {new_unit_name}")
@@ -173,6 +182,12 @@ async def test_backup_aws(ops_test: OpsTest, cloud_configs: Tuple[Dict, Dict]) -
         await action.wait()
         backups = action.results.get("backups")
         assert backups, "backups not outputted"
+
+        # Remove S3 relation to ensure "move to another cluster" blocked status is gone
+        await ops_test.model.applications[database_app_name].remove_relation(
+            f"{database_app_name}:s3-parameters", f"{S3_INTEGRATOR_APP_NAME}:s3-credentials"
+        )
+
         await ops_test.model.wait_for_idle(status="active", timeout=1000)
 
     # Remove the database app.
@@ -279,7 +294,7 @@ async def test_restore_on_new_cluster(ops_test: OpsTest, github_secrets) -> None
             database_app_name,
             0,
             S3_INTEGRATOR_APP_NAME,
-            ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE,
+            MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
         )
 
     # Check that the backup was correctly restored by having only the first created table.
diff --git a/tests/integration/test_backups_pitr.py b/tests/integration/test_backups_pitr.py
new file mode 100644
index 0000000000..8f60ab3ff9
--- /dev/null
+++ b/tests/integration/test_backups_pitr.py
@@ -0,0 +1,289 @@
+#!/usr/bin/env python3
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+import logging
+import uuid
+from typing import Dict, Tuple
+
+import boto3
+import pytest as pytest
+from pytest_operator.plugin import OpsTest
+from tenacity import Retrying, stop_after_attempt, wait_exponential
+
+from . import architecture
+from .helpers import (
+    DATABASE_APP_NAME,
+    MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
+    build_and_deploy,
+    construct_endpoint,
+    db_connect,
+    get_password,
+    get_primary,
+    get_unit_address,
+    scale_application,
+)
+from .juju_ import juju_major_version
+
+CANNOT_RESTORE_PITR = "cannot restore PITR, juju debug-log for details"
+ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE = "the S3 repository has backups from another cluster"
+FAILED_TO_ACCESS_CREATE_BUCKET_ERROR_MESSAGE = (
+    "failed to access/create the bucket, check your S3 settings"
+)
+FAILED_TO_INITIALIZE_STANZA_ERROR_MESSAGE = "failed to initialize stanza, check your S3 settings"
+S3_INTEGRATOR_APP_NAME = "s3-integrator"
+if juju_major_version < 3:
+    tls_certificates_app_name = "tls-certificates-operator"
+    if architecture.architecture == "arm64":
+        tls_channel = "legacy/edge"
+    else:
+        tls_channel = "legacy/stable"
+    tls_config = {"generate-self-signed-certificates": "true", "ca-common-name": "Test CA"}
+else:
+    tls_certificates_app_name = "self-signed-certificates"
+    if architecture.architecture == "arm64":
+        tls_channel = "latest/edge"
+    else:
+        tls_channel = "latest/stable"
+    tls_config = {"ca-common-name": "Test CA"}
+
+logger = logging.getLogger(__name__)
+
+AWS = "AWS"
+GCP = "GCP"
+
+
+@pytest.fixture(scope="module")
+async def cloud_configs(ops_test: OpsTest, github_secrets) -> None:
+    # Define some configurations and credentials.
+    configs = {
+        AWS: {
+            "endpoint": "https://s3.amazonaws.com",
+            "bucket": "data-charms-testing",
+            "path": f"/postgresql-k8s/{uuid.uuid1()}",
+            "region": "us-east-1",
+        },
+        GCP: {
+            "endpoint": "https://storage.googleapis.com",
+            "bucket": "data-charms-testing",
+            "path": f"/postgresql-k8s/{uuid.uuid1()}",
+            "region": "",
+        },
+    }
+    credentials = {
+        AWS: {
+            "access-key": github_secrets["AWS_ACCESS_KEY"],
+            "secret-key": github_secrets["AWS_SECRET_KEY"],
+        },
+        GCP: {
+            "access-key": github_secrets["GCP_ACCESS_KEY"],
+            "secret-key": github_secrets["GCP_SECRET_KEY"],
+        },
+    }
+    yield configs, credentials
+    # Delete the previously created objects.
+    logger.info("deleting the previously created backups")
+    for cloud, config in configs.items():
+        session = boto3.session.Session(
+            aws_access_key_id=credentials[cloud]["access-key"],
+            aws_secret_access_key=credentials[cloud]["secret-key"],
+            region_name=config["region"],
+        )
+        s3 = session.resource(
+            "s3", endpoint_url=construct_endpoint(config["endpoint"], config["region"])
+        )
+        bucket = s3.Bucket(config["bucket"])
+        # GCS doesn't support batch delete operation, so delete the objects one by one.
+        for bucket_object in bucket.objects.filter(Prefix=config["path"].lstrip("/")):
+            bucket_object.delete()
+
+
+@pytest.mark.group(1)
+@pytest.mark.abort_on_fail
+async def test_pitr_backup(ops_test: OpsTest, cloud_configs: Tuple[Dict, Dict]) -> None:
+    """Build and deploy two units of PostgreSQL in AWS and then test the backup and restore actions."""
+    config = cloud_configs[0][AWS]
+    credentials = cloud_configs[1][AWS]
+    cloud = AWS.lower()
+
+    # Deploy S3 Integrator and TLS Certificates Operator.
+    await ops_test.model.deploy(S3_INTEGRATOR_APP_NAME)
+    await ops_test.model.deploy(tls_certificates_app_name, config=tls_config, channel=tls_channel)
+    # Deploy and relate PostgreSQL to S3 integrator (one database app for each cloud for now
+    # as archivo_mode is disabled after restoring the backup) and to TLS Certificates Operator
+    # (to be able to create backups from replicas).
+    database_app_name = f"{DATABASE_APP_NAME}-{cloud}"
+    await build_and_deploy(ops_test, 2, database_app_name=database_app_name, wait_for_idle=False)
+
+    await ops_test.model.relate(database_app_name, tls_certificates_app_name)
+    async with ops_test.fast_forward(fast_interval="60s"):
+        await ops_test.model.wait_for_idle(
+            apps=[database_app_name], status="active", timeout=1000, raise_on_error=False
+        )
+    await ops_test.model.relate(database_app_name, S3_INTEGRATOR_APP_NAME)
+
+    # Configure and set access and secret keys.
+    logger.info(f"configuring S3 integrator for {cloud}")
+    await ops_test.model.applications[S3_INTEGRATOR_APP_NAME].set_config(config)
+    action = await ops_test.model.units.get(f"{S3_INTEGRATOR_APP_NAME}/0").run_action(
+        "sync-s3-credentials",
+        **credentials,
+    )
+    await action.wait()
+    async with ops_test.fast_forward(fast_interval="60s"):
+        await ops_test.model.wait_for_idle(
+            apps=[database_app_name, S3_INTEGRATOR_APP_NAME], status="active", timeout=1000
+        )
+
+    primary = await get_primary(ops_test, database_app_name)
+    for unit in ops_test.model.applications[database_app_name].units:
+        if unit.name != primary:
+            replica = unit.name
+            break
+
+    # Write some data.
+    password = await get_password(ops_test, database_app_name=database_app_name)
+    address = await get_unit_address(ops_test, primary)
+    logger.info("creating a table in the database")
+    with db_connect(host=address, password=password) as connection:
+        connection.autocommit = True
+        connection.cursor().execute(
+            "CREATE TABLE IF NOT EXISTS backup_table_1 (test_column INT );"
+        )
+    connection.close()
+
+    # With a stable cluster, Run the "create backup" action
+    async with ops_test.fast_forward():
+        await ops_test.model.wait_for_idle(status="active", timeout=1000, idle_period=30)
+    logger.info("creating a backup")
+    action = await ops_test.model.units.get(replica).run_action("create-backup")
+    await action.wait()
+    backup_status = action.results.get("backup-status")
+    assert backup_status, "backup hasn't succeeded"
+    async with ops_test.fast_forward():
+        await ops_test.model.wait_for_idle(status="active", timeout=1000)
+
+    # Run the "list backups" action.
+    logger.info("listing the available backups")
+    action = await ops_test.model.units.get(replica).run_action("list-backups")
+    await action.wait()
+    backups = action.results.get("backups")
+    # 5 lines for header output, 1 backup line ==> 6 total lines
+    assert len(backups.split("\n")) == 6, "full backup is not outputted"
+    await ops_test.model.wait_for_idle(status="active", timeout=1000)
+
+    # Write some data.
+    logger.info("creating after-backup data in the database")
+    with db_connect(host=address, password=password) as connection:
+        connection.autocommit = True
+        connection.cursor().execute(
+            "INSERT INTO backup_table_1 (test_column) VALUES (1), (2), (3), (4), (5);"
+        )
+    connection.close()
+    with db_connect(host=address, password=password) as connection, connection.cursor() as cursor:
+        cursor.execute("SELECT current_timestamp;")
+        after_backup_ts = str(cursor.fetchone()[0])
+    connection.close()
+    with db_connect(host=address, password=password) as connection:
+        connection.autocommit = True
+        connection.cursor().execute("CREATE TABLE IF NOT EXISTS backup_table_2 (test_column INT);")
+    connection.close()
+    with db_connect(host=address, password=password) as connection:
+        connection.autocommit = True
+        connection.cursor().execute("SELECT pg_switch_wal();")
+    connection.close()
+
+    async with ops_test.fast_forward(fast_interval="60s"):
+        await scale_application(ops_test, database_app_name, 1)
+    remaining_unit = ops_test.model.units.get(f"{database_app_name}/0")
+
+    most_recent_backup = backups.split("\n")[-1]
+    backup_id = most_recent_backup.split()[0]
+    # Wrong timestamp pointing to one year ahead
+    wrong_ts = after_backup_ts.replace(after_backup_ts[:4], str(int(after_backup_ts[:4]) + 1), 1)
+
+    # Run the "restore backup" action with bad PITR parameter.
+    logger.info("restoring the backup with bad restore-to-time parameter")
+    action = await ops_test.model.units.get(f"{database_app_name}/0").run_action(
+        "restore", **{"backup-id": backup_id, "restore-to-time": "bad data"}
+    )
+    await action.wait()
+    assert (
+        action.status == "failed"
+    ), "action must fail with bad restore-to-time parameter, but it succeeded"
+
+    # Run the "restore backup" action with unreachable PITR parameter.
+    logger.info("restoring the backup with unreachable restore-to-time parameter")
+    action = await ops_test.model.units.get(f"{database_app_name}/0").run_action(
+        "restore", **{"backup-id": backup_id, "restore-to-time": wrong_ts}
+    )
+    await action.wait()
+    logger.info("waiting for the database charm to become blocked")
+    async with ops_test.fast_forward():
+        await ops_test.model.block_until(
+            lambda: ops_test.model.units.get(f"{database_app_name}/0").workload_status_message
+            == CANNOT_RESTORE_PITR,
+            timeout=1000,
+        )
+    logger.info(
+        "database charm become in blocked state, as supposed to be with unreachable PITR parameter"
+    )
+
+    # Run the "restore backup" action.
+    for attempt in Retrying(
+        stop=stop_after_attempt(10), wait=wait_exponential(multiplier=1, min=2, max=30)
+    ):
+        with attempt:
+            logger.info("restoring the backup")
+            action = await remaining_unit.run_action(
+                "restore", **{"backup-id": backup_id, "restore-to-time": after_backup_ts}
+            )
+            await action.wait()
+            restore_status = action.results.get("restore-status")
+            assert restore_status, "restore hasn't succeeded"
+
+    # Wait for the restore to complete.
+    async with ops_test.fast_forward():
+        await ops_test.model.block_until(
+            lambda: remaining_unit.workload_status_message
+            == MOVE_RESTORED_CLUSTER_TO_ANOTHER_BUCKET,
+            timeout=1000,
+        )
+
+        # Check that the backup was correctly restored.
+        primary = await get_primary(ops_test, database_app_name)
+        address = await get_unit_address(ops_test, primary)
+        logger.info("checking that the backup was correctly restored")
+        with db_connect(
+            host=address, password=password
+        ) as connection, connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT EXISTS (SELECT FROM information_schema.tables"
+                " WHERE table_schema = 'public' AND table_name = 'backup_table_1');"
+            )
+            assert cursor.fetchone()[
+                0
+            ], "backup wasn't correctly restored: table 'backup_table_1' doesn't exist"
+            cursor.execute("SELECT COUNT(1) FROM backup_table_1;")
+            assert (
+                int(cursor.fetchone()[0]) == 5
+            ), "backup wasn't correctly restored: table 'backup_table_1' doesn't have 5 rows"
+            cursor.execute(
+                "SELECT EXISTS (SELECT FROM information_schema.tables"
+                " WHERE table_schema = 'public' AND table_name = 'backup_table_2');"
+            )
+            assert not cursor.fetchone()[
+                0
+            ], "backup wasn't correctly restored: table 'backup_table_2' exists"
+        connection.close()
+
+        # Remove S3 relation to ensure "move to another cluster" blocked status is gone
+        await ops_test.model.applications[database_app_name].remove_relation(
+            f"{database_app_name}:s3-parameters", f"{S3_INTEGRATOR_APP_NAME}:s3-credentials"
+        )
+
+        await ops_test.model.wait_for_idle(status="active", timeout=1000)
+
+    # Remove the database app.
+    await ops_test.model.remove_application(database_app_name, block_until_done=True)
+    # Remove the TLS operator.
+    await ops_test.model.remove_application(tls_certificates_app_name, block_until_done=True)
diff --git a/tests/unit/test_backups.py b/tests/unit/test_backups.py
index 032ae0abbf..41eb9a6440 100644
--- a/tests/unit/test_backups.py
+++ b/tests/unit/test_backups.py
@@ -210,6 +210,9 @@ def test_can_use_s3_repository(harness):
             new_callable=PropertyMock(return_value="14.10"),
         ) as _rock_postgresql_version,
         patch("charm.PostgreSQLBackups._execute_command") as _execute_command,
+        patch(
+            "charms.postgresql_k8s.v0.postgresql.PostgreSQL.get_last_archived_wal"
+        ) as _get_last_archived_wal,
     ):
         peer_rel_id = harness.model.get_relation(PEER).id
         # Define the stanza name inside the unit relation data.
@@ -1360,6 +1363,12 @@ def test_on_restore_action(harness):
         patch("charm.PostgreSQLBackups._list_backups") as _list_backups,
         patch("charm.PostgreSQLBackups._fetch_backup_from_id") as _fetch_backup_from_id,
         patch("charm.PostgreSQLBackups._pre_restore_checks") as _pre_restore_checks,
+        patch(
+            "charm.PostgresqlOperatorCharm.override_patroni_on_failure_condition"
+        ) as _override_patroni_on_failure_condition,
+        patch(
+            "charm.PostgresqlOperatorCharm.restore_patroni_on_failure_condition"
+        ) as _restore_patroni_on_failure_condition,
     ):
         peer_rel_id = harness.model.get_relation(PEER).id
         # Test when pre restore checks fail.
@@ -1468,6 +1477,7 @@ def test_on_restore_action(harness):
         assert harness.get_relation_data(peer_rel_id, harness.charm.app) == {
             "restoring-backup": "20230101-090000F",
             "restore-stanza": f"{harness.charm.model.name}.{harness.charm.cluster_name}",
+            "require-change-bucket-after-restore": "True",
         }
         _create_pgdata.assert_called_once()
         _update_config.assert_called_once()
diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py
index 16d89ffb04..d00edef8fe 100644
--- a/tests/unit/test_charm.py
+++ b/tests/unit/test_charm.py
@@ -883,6 +883,7 @@ def test_postgresql_layer(harness):
                     "summary": "entrypoint of the postgresql + patroni image",
                     "command": "patroni /var/lib/postgresql/data/patroni.yml",
                     "startup": "enabled",
+                    "on-failure": "restart",
                     "user": "postgres",
                     "group": "postgres",
                     "environment": {
@@ -1630,6 +1631,9 @@ def test_update_config(harness):
             backup_id=None,
             stanza=None,
             restore_stanza=None,
+            pitr_target=None,
+            restore_to_latest=False,
+            disable_pgbackrest_archiving=False,
             parameters={"test": "test"},
         )
         _handle_postgresql_restart_need.assert_called_once()
@@ -1651,6 +1655,9 @@ def test_update_config(harness):
             backup_id=None,
             stanza=None,
             restore_stanza=None,
+            pitr_target=None,
+            restore_to_latest=False,
+            disable_pgbackrest_archiving=False,
             parameters={"test": "test"},
         )
         _handle_postgresql_restart_need.assert_called_once()