Copy additional backup locations in jobs (#5890)

Instead of copying the backup in the main job, lets copy them in
separate job per location. This allows to use the same backup error
handling mechanism as for add-ons and folders.

This makes the stage introduced in #5784 somewhat redundant, but
before removing it, let's see if this approach works out.

supervisor/backups/manager.py

@@ -378,66 +378,69 @@ class BackupManager(FileConfiguration, JobGroup):
         if not backup.all_locations:
             del self._backups[backup.slug]
 
+    @Job(name="backup_copy_to_location", cleanup=False)
+    async def _copy_to_location(
+        self, backup: Backup, location: LOCATION_TYPE
+    ) -> tuple[str | None, Path]:
+        """Copy a backup file to the default location."""
+        location_name = location.name if isinstance(location, Mount) else location
+        self.sys_jobs.current.reference = location_name
+        try:
+            if location == LOCATION_CLOUD_BACKUP:
+                destination = self.sys_config.path_core_backup
+            elif location:
+                location_mount = cast(Mount, location)
+                if not location_mount.local_where.is_mount():
+                    raise BackupMountDownError(
+                        f"{location_mount.name} is down, cannot copy to it",
+                        _LOGGER.error,
+                    )
+                destination = location_mount.local_where
+            else:
+                destination = self.sys_config.path_backup
+
+            path = await self.sys_run_in_executor(copy, backup.tarfile, destination)
+            return (location_name, Path(path))
+        except OSError as err:
+            msg = f"Could not copy backup to {location_name} due to: {err!s}"
+
+            if err.errno == errno.EBADMSG and location in {
+                LOCATION_CLOUD_BACKUP,
+                None,
+            }:
+                raise BackupDataDiskBadMessageError(msg, _LOGGER.error) from err
+            raise BackupError(msg, _LOGGER.error) from err
+
+    @Job(name="backup_copy_to_additional_locations", cleanup=False)
     async def _copy_to_additional_locations(
         self,
         backup: Backup,
         locations: list[LOCATION_TYPE],
     ):
         """Copy a backup file to additional locations."""
-
         all_new_locations: dict[str | None, Path] = {}
+        for location in locations:
+            try:
+                location_name, path = await self._copy_to_location(backup, location)
+                all_new_locations[location_name] = path
+            except BackupDataDiskBadMessageError as err:
+                self.sys_resolution.add_unhealthy_reason(
+                    UnhealthyReason.OSERROR_BAD_MESSAGE
+                )
+                self.sys_jobs.current.capture_error(err)
+            except BackupError as err:
+                self.sys_jobs.current.capture_error(err)
 
-        def copy_to_additional_locations() -> None:
+        backup.all_locations.update(
-            """Copy backup file to additional locations."""
+            {
-            nonlocal all_new_locations
+                loc: BackupLocation(
-            for location in locations:
+                    path=path,
-                try:
+                    protected=backup.protected,
-                    if location == LOCATION_CLOUD_BACKUP:
+                    size_bytes=backup.size_bytes,
-                        all_new_locations[LOCATION_CLOUD_BACKUP] = Path(
+                )
-                            copy(backup.tarfile, self.sys_config.path_core_backup)
+                for loc, path in all_new_locations.items()
-                        )
+            }
-                    elif location:
+        )
-                        location_mount = cast(Mount, location)
-                        if not location_mount.local_where.is_mount():
-                            raise BackupMountDownError(
-                                f"{location_mount.name} is down, cannot copy to it",
-                                _LOGGER.error,
-                            )
-                        all_new_locations[location_mount.name] = Path(
-                            copy(backup.tarfile, location_mount.local_where)
-                        )
-                    else:
-                        all_new_locations[None] = Path(
-                            copy(backup.tarfile, self.sys_config.path_backup)
-                        )
-                except OSError as err:
-                    msg = f"Could not copy backup to {location.name if isinstance(location, Mount) else location} due to: {err!s}"
-
-                    if err.errno == errno.EBADMSG and location in {
-                        LOCATION_CLOUD_BACKUP,
-                        None,
-                    }:
-                        raise BackupDataDiskBadMessageError(msg, _LOGGER.error) from err
-                    raise BackupError(msg, _LOGGER.error) from err
-
-        try:
-            await self.sys_run_in_executor(copy_to_additional_locations)
-        except BackupDataDiskBadMessageError:
-            self.sys_resolution.add_unhealthy_reason(
-                UnhealthyReason.OSERROR_BAD_MESSAGE
-            )
-            raise
-        finally:
-            backup.all_locations.update(
-                {
-                    loc: BackupLocation(
-                        path=path,
-                        protected=backup.protected,
-                        size_bytes=backup.size_bytes,
-                    )
-                    for loc, path in all_new_locations.items()
-                }
-            )
 
     @Job(name="backup_manager_import_backup")
     async def import_backup(
@@ -566,12 +569,7 @@ class BackupManager(FileConfiguration, JobGroup):
 
             if additional_locations:
                 self._change_stage(BackupJobStage.COPY_ADDITONAL_LOCATIONS, backup)
-                try:
+                await self._copy_to_additional_locations(backup, additional_locations)
-                    await self._copy_to_additional_locations(
-                        backup, additional_locations
-                    )
-                except BackupError as err:
-                    self.sys_jobs.current.capture_error(err)
 
             if addon_start_tasks:
                 self._change_stage(BackupJobStage.AWAIT_ADDON_RESTARTS, backup)

tests/api/test_backups.py

@@ -729,11 +729,26 @@ async def test_backup_to_multiple_locations_error_on_copy(
     assert result["data"]["jobs"][0]["name"] == f"backup_manager_{backup_type}_backup"
     assert result["data"]["jobs"][0]["reference"] == slug
     assert result["data"]["jobs"][0]["done"] is True
-    assert result["data"]["jobs"][0]["errors"] == [
+    assert len(result["data"]["jobs"][0]["errors"]) == 0
+
+    # Errors during copy to additional location should be recorded in child jobs only.
+    assert (
+        result["data"]["jobs"][0]["child_jobs"][-1]["name"]
+        == "backup_copy_to_additional_locations"
+    )
+    assert (
+        result["data"]["jobs"][0]["child_jobs"][-1]["child_jobs"][0]["name"]
+        == "backup_copy_to_location"
+    )
+    assert (
+        result["data"]["jobs"][0]["child_jobs"][-1]["child_jobs"][0]["reference"]
+        == ".cloud_backup"
+    )
+    assert result["data"]["jobs"][0]["child_jobs"][-1]["child_jobs"][0]["errors"] == [
         {
             "type": "BackupError",
             "message": "Could not copy backup to .cloud_backup due to: ",
-            "stage": "copy_additional_locations",
+            "stage": None,
         }
     ]