rollback lock file when installer fails

src/poetry/installation/installer.py

@@ -1,5 +1,9 @@
 from __future__ import annotations
 
+import shutil
+import tempfile
+
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 from cleo.io.null_io import NullIO
@@ -60,6 +64,7 @@ def __init__(
 
         self._execute_operations = True
         self._lock = False
+        self._lock_prev_state: Path | None = None
 
         self._whitelist: list[NormalizedName] = []
 
@@ -282,7 +287,7 @@ def _do_install(self) -> int:
         self._populate_lockfile_repo(lockfile_repo, ops)
 
         if self._update:
-            self._write_lock_file(lockfile_repo)
+            self._write_lock_file(lockfile_repo, backup=not self._lock)
 
             if self._lock:
                 # If we are only in lock mode, no need to go any further
@@ -352,10 +357,22 @@ def _do_install(self) -> int:
         self._filter_operations(ops, lockfile_repo)
 
         # Execute operations
-        return self._execute(ops)
+        status = self._execute(ops)
+
+        # On errors, restore lock file state
+        if status != 0:
+            self._restore_lock_file()
+
+        self._cleanup()
 
-    def _write_lock_file(self, repo: LockfileRepository, force: bool = False) -> None:
+        return status
+
+    def _write_lock_file(
+        self, repo: LockfileRepository, force: bool = False, backup: bool = False
+    ) -> None:
         if self._write_lock and (force or self._update):
+            if backup:
+                self._backup_lock_file()
             updated_lock = self._locker.set_lock_data(self._package, repo.packages)
 
             if updated_lock:
@@ -567,3 +584,30 @@ def _get_installer(self) -> BaseInstaller:
 
     def _get_installed(self) -> InstalledRepository:
         return InstalledRepository.load(self._env)
+
+    def _backup_lock_file(self) -> None:
+        # Store current state of the lock file as a backup if the lock file exists
+        if not self._locker.lock.exists():
+            self._lock_prev_state = Path("NOT_EXIST")
+        else:
+            self._lock_prev_state = Path(tempfile.mkstemp("-poetry.lock", "backup-")[1])
+            shutil.copyfile(self._locker.lock, self._lock_prev_state)
+
+    def _restore_lock_file(self) -> None:
+        # Restore last state of the lock file, deleting it if there was none before.
+        if self._lock_prev_state:
+            if self._lock_prev_state == Path("NOT_EXIST"):
+                self._locker.lock.unlink(missing_ok=True)
+            else:
+                self._lock_prev_state.replace(self._locker.lock)
+
+        # Enforce to refresh lock data from the restored file
+        del self._locker.lock_data
+
+    def _cleanup(self) -> None:
+        if self._lock_prev_state:
+            self._lock_prev_state.unlink(missing_ok=True)
+            self._lock_prev_state = None
+
+    def __del__(self) -> None:
+        self._cleanup()

src/poetry/packages/locker.py

@@ -70,6 +70,10 @@ def lock_data(self) -> dict[str, Any]:
 
         return self._lock_data
 
+    @lock_data.deleter
+    def lock_data(self) -> None:
+        self._lock_data = None
+
     def is_locked(self) -> bool:
         """
         Checks whether the locker has been locked (lockfile found).