[Refinement] boot_control.grub: improve robustness against boot load/init, split out ota_status control and slot mount logic

otaclient/app/boot_control/_common.py

@@ -424,8 +424,7 @@ def reboot(cls):
 
 
 class OTAStatusFilesControl:
-    """Logics for controlling otaclient's behavior using ota_status files,
-    including status, slot_in_use and version.
+    """Logics for controlling otaclient's OTA status with corresponding files.
 
     OTAStatus files:
         status: current slot's OTA status
@@ -444,56 +443,38 @@ def __init__(
         current_ota_status_dir: Union[str, Path],
         standby_ota_status_dir: Union[str, Path],
         finalize_switching_boot: FinalizeSwitchBootFunc,
+        force_initialize: bool = False,
     ) -> None:
         self.active_slot = active_slot
         self.standby_slot = standby_slot
         self.current_ota_status_dir = Path(current_ota_status_dir)
         self.standby_ota_status_dir = Path(standby_ota_status_dir)
         self.finalize_switching_boot = finalize_switching_boot
 
-        # NOTE: pre-assign live ota_status with the loaded ota_status,
-        #       and then update live ota_status in below.
-        #       The reason is for some platform, like raspberry pi 4B,
-        #       the finalize_switching_boot might be slow, so we first provide
-        #       live ota_status the same as loaded ota_status(or INITIALIZED),
-        #       then update it after the init_ota_status_files finished.
-        _loaded_ota_status = self._load_current_status()
-        self._ota_status = (
-            _loaded_ota_status if _loaded_ota_status else wrapper.StatusOta.INITIALIZED
-        )
-
-        _loaded_slot_in_use = self._load_current_slot_in_use()
-        if _loaded_slot_in_use and _loaded_slot_in_use != self.active_slot:
-            logger.warning(
-                f"boot into old slot {self.active_slot}, "
-                f"but slot_in_use indicates it should boot into {_loaded_slot_in_use}, "
-                "this might indicate a failed finalization at first reboot after update/rollback"
-            )
-
-        # initializing ota_status control
-        self._init_ota_status_files()
+        self._force_initialize = force_initialize
+        self.current_ota_status_dir.mkdir(exist_ok=True, parents=True)
+        self._load_slot_in_use_file()
+        self._load_status_file()
         logger.info(
-            f"ota_status files parsing completed, ota_status is {self._ota_status}"
+            f"ota_status files parsing completed, ota_status is {self._ota_status.name}"
         )
 
-    def _init_ota_status_files(self):
+    def _load_status_file(self):
         """Check and/or init ota_status files for current slot."""
-        self.current_ota_status_dir.mkdir(exist_ok=True, parents=True)
-
-        # load ota_status and slot_in_use file
         _loaded_ota_status = self._load_current_status()
-        _loaded_slot_in_use = self._load_current_slot_in_use()
+        if self._force_initialize:
+            _loaded_ota_status = None
 
         # initialize ota_status files if not presented/incompleted/invalid
-        if not (_loaded_ota_status and _loaded_slot_in_use):
+        if not _loaded_ota_status:
             logger.info(
                 "ota_status files incompleted/not presented, "
-                "initializing and set/store status to INITIALIZED..."
+                f"initializing and set/store status to {wrapper.StatusOta.INITIALIZED.name}..."
             )
-            self._store_current_slot_in_use(self.active_slot)
             self._store_current_status(wrapper.StatusOta.INITIALIZED)
             self._ota_status = wrapper.StatusOta.INITIALIZED
             return
+        logger.info(f"status loaded from file: {_loaded_ota_status.name}")
 
         # status except UPDATING and ROLLBACKING(like SUCCESS/FAILURE/ROLLBACK_FAILURE)
         # are remained as it
@@ -505,8 +486,11 @@ def _init_ota_status_files(self):
             return
 
         # updating or rollbacking,
+        # NOTE: pre-assign live ota_status with the loaded ota_status before entering finalizing_switch_boot,
+        #       as some of the platform might have slow finalizing process(like raspberry).
+        self._ota_status = _loaded_ota_status
         # if is_switching_boot, execute the injected finalize_switching_boot function from
-        #   boot controller, transit the ota_status according to the execution result.
+        # boot controller, transit the ota_status according to the execution result.
         # NOTE(20230614): for boot controller during multi-stage reboot(like rpi_boot),
         #                 calling finalize_switching_boot might result in direct reboot,
         #                 in such case, otaclient will terminate and ota_status will not be updated.
@@ -528,7 +512,7 @@ def _init_ota_status_files(self):
         else:
             logger.error(
                 f"we are in {_loaded_ota_status.name} ota_status, "
-                f"but {_loaded_slot_in_use=} doesn't match {self.active_slot=}, "
+                "but ota_status files indicate that we are not in switching boot mode, "
                 "this indicates a failed first reboot"
             )
             self._ota_status = (
@@ -538,6 +522,27 @@ def _init_ota_status_files(self):
             )
             self._store_current_status(self._ota_status)
 
+    def _load_slot_in_use_file(self):
+        _loaded_slot_in_use = self._load_current_slot_in_use()
+        if self._force_initialize:
+            _loaded_slot_in_use = None
+
+        if not _loaded_slot_in_use:
+            # NOTE(20230831): this can also resolve the backward compatibility issue
+            #                 in is_switching_boot method when old otaclient doesn't create
+            #                 slot_in_use file.
+            self._store_current_slot_in_use(self.active_slot)
+            return
+        logger.info(f"slot_in_use loaded from file: {_loaded_slot_in_use}")
+
+        # check potential failed switching boot
+        if _loaded_slot_in_use and _loaded_slot_in_use != self.active_slot:
+            logger.warning(
+                f"boot into old slot {self.active_slot}, "
+                f"but slot_in_use indicates it should boot into {_loaded_slot_in_use}, "
+                "this might indicate a failed finalization at first reboot after update/rollback"
+            )
+
     # slot_in_use control
 
     def _store_current_slot_in_use(self, _slot: str):
@@ -637,15 +642,11 @@ def pre_rollback_standby(self):
         self._store_standby_status(wrapper.StatusOta.ROLLBACKING)
 
     def load_active_slot_version(self) -> str:
-        _version = read_str_from_file(
+        return read_str_from_file(
             self.current_ota_status_dir / cfg.OTA_VERSION_FNAME,
             missing_ok=True,
-            default="",
+            default=cfg.DEFAULT_VERSION_STR,
         )
-        if not _version:
-            logger.warning("version file not found, return empty version string")
-
-        return _version
 
     def on_failure(self):
         """Store FAILURE to status file on failure."""
@@ -655,8 +656,14 @@ def on_failure(self):
             self._store_standby_status(wrapper.StatusOta.FAILURE)
 
     @property
-    def ota_status(self) -> wrapper.StatusOta:
-        """Read only ota_status property."""
+    def booted_ota_status(self) -> wrapper.StatusOta:
+        """Loaded current slot's ota_status during boot control starts.
+
+        NOTE: distinguish between the live ota_status maintained by otaclient.
+
+        This property is only meant to be used once when otaclient starts up,
+        switch to use live_ota_status by otaclient after otaclient is running.
+        """
         return self._ota_status
 
 
@@ -705,7 +712,7 @@ def _load_current_ota_status(self) -> Optional[wrapper.StatusOta]:
             except KeyError:
                 pass  # invalid status string
 
-    def get_ota_status(self) -> wrapper.StatusOta:
+    def get_booted_ota_status(self) -> wrapper.StatusOta:
         return self.ota_status
 
 
@@ -788,7 +795,12 @@ def __init__(
         self.standby_slot_mount_point.mkdir(exist_ok=True, parents=True)
         self.active_slot_mount_point.mkdir(exist_ok=True, parents=True)
         # standby slot /boot dir
-        self.standby_boot_dir = self.standby_slot_mount_point / "boot"
+        # NOTE(20230907): this will always be <standby_slot_mp>/boot,
+        #                 in the future this attribute will not be used by
+        #                 standby slot creater.
+        self.standby_boot_dir = self.standby_slot_mount_point / Path(
+            cfg.BOOT_DIR
+        ).relative_to("/")
 
     def mount_standby(self, *, raise_exc: bool = True) -> bool:
         """Mount standby slot dev to <standby_slot_mount_point>.