diff options
Diffstat (limited to 'py_modules')
| -rw-r--r-- | py_modules/lsfg_vk/flatpak_service.py | 109 | ||||
| -rw-r--r-- | py_modules/lsfg_vk/installation.py | 76 |
2 files changed, 168 insertions, 17 deletions
diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index fe965aa..8410496 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -91,6 +91,52 @@ class FlatpakService(BaseService): plugin_dir = Path(__file__).resolve().parent.parent.parent return plugin_dir / BIN_DIR / filename + def _get_extension_id(self, version: str) -> str: + """Return the full user-installation ref for a supported runtime branch.""" + extension_ids = { + "23.08": self.extension_id_23_08, + "24.08": self.extension_id_24_08, + "25.08": self.extension_id_25_08, + } + try: + return extension_ids[version] + except KeyError as error: + raise ValueError(f"Unsupported Flatpak runtime version: {version}") from error + + def _get_installed_extension_commit(self, version: str) -> str: + """Record the deployed commit so a failed multi-runtime migration can roll back.""" + result = self._run_flatpak_command( + ["info", "--user", "--show-commit", self._get_extension_id(version)], + capture_output=True, + text=True, + check=True, + ) + commit = str(result.stdout or "").strip() + if not commit: + raise OSError(f"Flatpak did not report a deployed commit for {version}") + return commit + + def _rollback_extension(self, version: str, commit: str) -> Optional[str]: + """Restore one runtime to its previous deployed commit; return an error if it fails.""" + try: + result = self._run_flatpak_command( + [ + "update", + "--user", + "--assumeyes", + "--noninteractive", + f"--commit={commit}", + self._get_extension_id(version), + ], + capture_output=True, + text=True, + ) + except (OSError, subprocess.SubprocessError) as error: + return str(error) + if result.returncode != 0: + return result.stderr or f"Flatpak rollback failed for {version}" + return None + def _get_installed_extension_versions(self) -> list[str]: """Return installed lsfg-vk runtime branches from the user Flatpak installation.""" result = self._run_flatpak_command( @@ -271,7 +317,7 @@ class FlatpakService(BaseService): def _get_flatpak_app_ids(self) -> list[str]: """Return application IDs in the user Flatpak installation.""" result = self._run_flatpak_command( - ["list", "--app"], + ["list", "--user", "--app"], capture_output=True, text=True, check=True, @@ -348,6 +394,8 @@ class FlatpakService(BaseService): empty_result = { "updated_versions": [], "failed_versions": [], + "rolled_back_versions": [], + "rollback_failed_versions": [], "migrated_apps": [], "failed_apps": [], } @@ -369,6 +417,37 @@ class FlatpakService(BaseService): **empty_result, ) + missing_bundles = [ + {"version": version, "error": f"Bundled Flatpak extension not found for {version}"} + for version in installed_versions + if not self._get_bundled_extension_path(version).is_file() + ] + if missing_bundles: + return self._error_response( + BaseResponse, + "Flatpak runtime migration was not started because a bundled asset is missing", + skipped=False, + failed_versions=missing_bundles, + **{key: value for key, value in empty_result.items() if key != "failed_versions"}, + ) + + previous_commits = {} + commit_failures = [] + for version in installed_versions: + try: + previous_commits[version] = self._get_installed_extension_commit(version) + except (subprocess.CalledProcessError, OSError) as error: + stderr = getattr(error, "stderr", None) + commit_failures.append({"version": version, "error": stderr or str(error)}) + if commit_failures: + return self._error_response( + BaseResponse, + "Flatpak runtime migration was not started because existing commits could not be recorded", + skipped=False, + failed_versions=commit_failures, + **{key: value for key, value in empty_result.items() if key != "failed_versions"}, + ) + updated_versions = [] failed_versions = [] for version in installed_versions: @@ -379,12 +458,29 @@ class FlatpakService(BaseService): failed_versions.append({"version": version, "error": result.get("error", "unknown error")}) if failed_versions: + rolled_back_versions = [] + rollback_failed_versions = [] + for version in reversed(updated_versions): + rollback_error = self._rollback_extension(version, previous_commits[version]) + if rollback_error is None: + rolled_back_versions.append(version) + else: + rollback_failed_versions.append({"version": version, "error": rollback_error}) + + rollback_message = ( + f"rolled back {len(rolled_back_versions)} runtime(s)" + if not rollback_failed_versions + else f"could not roll back {len(rollback_failed_versions)} runtime(s)" + ) return self._error_response( BaseResponse, - "Flatpak runtime migration failed; legacy app overrides were left unchanged", + f"Flatpak runtime migration failed; {rollback_message}; " + "legacy app overrides were left unchanged", skipped=False, updated_versions=updated_versions, failed_versions=failed_versions, + rolled_back_versions=rolled_back_versions, + rollback_failed_versions=rollback_failed_versions, migrated_apps=[], failed_apps=[], ) @@ -393,6 +489,8 @@ class FlatpakService(BaseService): response = { "updated_versions": updated_versions, "failed_versions": [], + "rolled_back_versions": [], + "rollback_failed_versions": [], "migrated_apps": override_result.get("migrated_apps", []), "failed_apps": override_result.get("failed_apps", []), } @@ -428,12 +526,7 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): return self._error_response(BaseResponse, "Flatpak is not available on this system") - if version == "23.08": - extension_id = self.extension_id_23_08 - elif version == "24.08": - extension_id = self.extension_id_24_08 - else: - extension_id = self.extension_id_25_08 + extension_id = self._get_extension_id(version) result = self._run_flatpak_command( ["uninstall", "--user", "--noninteractive", extension_id], diff --git a/py_modules/lsfg_vk/installation.py b/py_modules/lsfg_vk/installation.py index 7eee848..1ba091a 100644 --- a/py_modules/lsfg_vk/installation.py +++ b/py_modules/lsfg_vk/installation.py @@ -45,10 +45,16 @@ class InstallationService(BaseService): return self._error_response(InstallationResponse, error_msg, message="") self._ensure_directories() - self._extract_and_install_files(archive_path) - profile_data = self._create_config_file() - self._create_lsfg_launch_script(profile_data) - self._remove_legacy_layer_files() + with tempfile.TemporaryDirectory(prefix="lsfg-vk-install-") as transaction_dir: + snapshots = self._snapshot_installation_files(Path(transaction_dir)) + try: + self._extract_and_install_files(archive_path) + profile_data = self._create_config_file() + self._create_lsfg_launch_script(profile_data) + self._remove_legacy_layer_files() + except Exception: + self._restore_installation_files(snapshots) + raise message = "lsfg-vk v2 installed successfully" if self._config_recovery_backup is not None: @@ -126,12 +132,54 @@ class InstallationService(BaseService): """Point the manifest at ~/.local/lib after Decky installs it.""" try: data = json.loads(source.read_text(encoding="utf-8")) + if not isinstance(data, dict) or not isinstance(data.get("layer"), dict): + raise ValueError("manifest is missing its layer object") data["layer"]["library_path"] = "../../../lib/liblsfg-vk-layer.so" - destination.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") - destination.chmod(0o644) - except (json.JSONDecodeError, KeyError, OSError) as error: + self._write_file(destination, json.dumps(data, indent=2) + "\n", 0o644) + except (json.JSONDecodeError, OSError, TypeError, UnicodeError, ValueError) as error: self.log.error("Error fixing JSON file %s: %s", source, error) - self._copy_plugin_file(source, destination) + raise OSError(f"Invalid Vulkan layer manifest {source}: {error}") from error + + def _installation_paths(self) -> tuple[Path, ...]: + """Return every file that can be changed by a native installation.""" + return ( + self.lib_file, + self.json_file, + self.config_file_path, + self.lsfg_launch_script_path, + self.legacy_lib_file, + self.legacy_json_file, + self.config_file_path.with_name(f"{self.config_file_path.name}.v1.bak"), + self.config_file_path.with_name(f"{self.config_file_path.name}.unrecognized.bak"), + ) + + def _snapshot_installation_files(self, transaction_dir: Path) -> Dict[Path, Path | None]: + """Snapshot installation files so a failed update can be rolled back.""" + snapshots: Dict[Path, Path | None] = {} + for index, path in enumerate(self._installation_paths()): + if not path.exists() and not path.is_symlink(): + snapshots[path] = None + continue + if not path.is_file(): + raise OSError(f"Installation target is not a regular file: {path}") + backup = transaction_dir / str(index) + shutil.copy2(path, backup) + snapshots[path] = backup + return snapshots + + def _restore_installation_files(self, snapshots: Dict[Path, Path | None]) -> None: + """Restore a previous installation after a failed update.""" + try: + for path, backup in snapshots.items(): + if path.exists() or path.is_symlink(): + path.unlink() + if backup is not None: + path.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(backup, path) + self.log.warning("Rolled back incomplete lsfg-vk installation") + except OSError as error: + self.log.error("Could not fully roll back lsfg-vk installation: %s", error) + raise def _create_config_file(self) -> ProfileData: """Migrate v1 once, normalize v2, and retain Decky launch workarounds.""" @@ -222,7 +270,17 @@ class InstallationService(BaseService): def _merge_config_with_defaults(self, existing: ProfileData, dll_service) -> ProfileData: defaults = ConfigurationManager.get_defaults_with_dll_detection(dll_service) global_config = dict(existing.get("global_config", {})) - global_config.setdefault("dll", defaults["dll"]) + configured_dll = str(global_config.get("dll", "") or "").strip() + detected_dll = str(defaults.get("dll", "") or "").strip() + if configured_dll and Path(configured_dll).is_file(): + global_config["dll"] = configured_dll + elif detected_dll and Path(detected_dll).is_file(): + self.log.warning("Replacing stale configured Lossless.dll path %s with %s", configured_dll, detected_dll) + global_config["dll"] = detected_dll + else: + if configured_dll: + self.log.warning("Clearing stale configured Lossless.dll path %s", configured_dll) + global_config["dll"] = "" global_config.setdefault("allow_fp16", defaults["allow_fp16"]) profiles: Dict[str, Any] = {} |