Simplify runtime journal recovery

Author: tobiasraabe

src/_pytask/journal.py

@@ -28,7 +28,7 @@ def append(self, payload: msgspec.Struct) -> None:
             journal_file.write(msgspec.json.encode(payload) + b"\n")
 
     def read(self) -> list[T]:
-        """Read entries, stopping at the first invalid line."""
+        """Read entries, clearing the journal on decode errors."""
         if not self.path.exists():
             return []
 
@@ -39,7 +39,8 @@ def read(self) -> list[T]:
             try:
                 entries.append(msgspec.json.decode(line, type=self.type_))
             except msgspec.DecodeError:
-                break
+                self.delete()
+                return []
         return entries
 
     def delete(self) -> None:

src/_pytask/runtime_store.py

@@ -7,7 +7,6 @@
 from typing import TYPE_CHECKING
 
 import msgspec
-from packaging.version import Version
 
 from _pytask.journal import JsonlJournal
 
@@ -19,14 +18,6 @@
 CURRENT_RUNTIME_VERSION = "1"
 
 
-class RuntimeStoreError(Exception):
-    """Raised when reading or writing runtime files fails."""
-
-
-class RuntimeStoreVersionError(RuntimeStoreError):
-    """Raised when a runtime file version is not supported."""
-
-
 class _RuntimeEntry(msgspec.Struct):
     id: str
     date: float
@@ -63,15 +54,12 @@ def _read_runtimes(path: Path) -> _RuntimeFile | None:
     try:
         data = msgspec.json.decode(path.read_bytes(), type=_RuntimeFile)
     except msgspec.DecodeError:
-        msg = "Runtime file has invalid format."
-        raise RuntimeStoreError(msg) from None
+        path.unlink()
+        return None
 
-    if Version(data.runtime_version) != Version(CURRENT_RUNTIME_VERSION):
-        msg = (
-            f"Unsupported runtime-version {data.runtime_version!r}. "
-            f"Current version is {CURRENT_RUNTIME_VERSION}."
-        )
-        raise RuntimeStoreVersionError(msg)
+    if data.runtime_version != CURRENT_RUNTIME_VERSION:
+        path.unlink()
+        return None
     return data
 
 
@@ -87,12 +75,9 @@ def _read_journal(
 ) -> list[_RuntimeJournalEntry]:
     entries = journal.read()
     for entry in entries:
-        if Version(entry.runtime_version) != Version(CURRENT_RUNTIME_VERSION):
-            msg = (
-                f"Unsupported runtime-version {entry.runtime_version!r}. "
-                f"Current version is {CURRENT_RUNTIME_VERSION}."
-            )
-            raise RuntimeStoreVersionError(msg)
+        if entry.runtime_version != CURRENT_RUNTIME_VERSION:
+            journal.delete()
+            return []
     return entries
 
 
@@ -147,12 +132,8 @@ def from_root(cls, root: Path) -> RuntimeState:
     def _rebuild_index(self) -> None:
         self._index = {entry.id: entry for entry in self.runtimes.task}
 
-    @staticmethod
-    def _task_id(task: PTask) -> str:
-        return task.name
-
     def update_task(self, task: PTask, start: float, end: float) -> None:
-        task_id = self._task_id(task)
+        task_id = task.name
         entry = _RuntimeEntry(id=task_id, date=start, duration=end - start)
         self._index[entry.id] = entry
         self.runtimes = _RuntimeFile(
@@ -170,7 +151,7 @@ def update_task(self, task: PTask, start: float, end: float) -> None:
         self._dirty = True
 
     def get_duration(self, task: PTask) -> float | None:
-        task_id = self._task_id(task)
+        task_id = task.name
         entry = self._index.get(task_id)
         if entry is None:
             return None