fix: wal replay recovery

benchmarks/test_wal.py

@@ -0,0 +1,84 @@
+"""Benchmarks for write-ahead log persistence."""
+
+from pathlib import Path
+
+import pytest
+
+from seriousdb.wal import SetEntry, WriteAheadLog
+
+from ._support import WARMUP_ROUNDS
+
+
+@pytest.mark.benchmark(group="wal")
+def test_wal_append(
+    benchmark,
+    database_file: Path,
+    measured_rounds: int,
+) -> None:
+    """Measure appending one entry to the WAL, including fsync."""
+
+    entry = SetEntry(key="key", value="value")
+
+    def prepare_wal():
+        database_file.unlink(missing_ok=True)
+        wal = WriteAheadLog(str(database_file))
+        return (wal,), {}
+
+    def append_entry(wal: WriteAheadLog) -> None:
+        wal.append(entry)
+
+    def close_wal(wal: WriteAheadLog) -> None:
+        wal.close()
+
+    benchmark.extra_info.update(
+        operation="WAL append",
+        persistence="write + flush + fsync",
+        key_bytes=len(entry.key.encode()),
+        value_bytes=len(entry.value.encode()),
+    )
+
+    benchmark.pedantic(
+        append_entry,
+        setup=prepare_wal,
+        teardown=close_wal,
+        rounds=50,
+        warmup_rounds=WARMUP_ROUNDS,
+    )
+
+
+@pytest.mark.benchmark(group="wal")
+def test_wal_append_many(
+    benchmark,
+    database_file: Path,
+) -> None:
+    """Measure repeated appends using one open WAL."""
+
+    entry = SetEntry(key="key", value="value")
+
+    def prepare_wal():
+        database_file.unlink(missing_ok=True)
+        wal = WriteAheadLog(str(database_file))
+        return (wal,), {}
+
+    def append_many(wal: WriteAheadLog) -> None:
+        for _ in range(1000):
+            wal.append(entry)
+
+    def close_wal(wal: WriteAheadLog) -> None:
+        wal.close()
+
+    benchmark.extra_info.update(
+        operation="1000 WAL appends",
+        persistence="write + flush + fsync",
+        entries=1000,
+        key_bytes=len(entry.key.encode()),
+        value_bytes=len(entry.value.encode()),
+    )
+
+    benchmark.pedantic(
+        append_many,
+        setup=prepare_wal,
+        teardown=close_wal,
+        rounds=10,
+        warmup_rounds=WARMUP_ROUNDS,
+    )

src/seriousdb/cache.py

@@ -320,15 +320,23 @@ def load(self, filename: str) -> None:
                     db = _write_default(filename)
 
             wal = WriteAheadLog(f"{filename}.wal")
-            replayed = wal.replay()
-
-            for entry in replayed:
-                entry.apply(db)
-
+            try:
+                replayed = wal.replay()
+                for entry in replayed:
+                    entry.apply(db)
+            except Exception:
+                try:
+                    wal.close()
+                except OSError:
+                    pass
+                raise
+            old_wal = self.wal
             self.db = db
             self.filename = filename
             self.wal = wal
             self._writes_since_compact = len(replayed)
+            if old_wal is not None:
+                old_wal.close()
 
     def flush(self) -> None:
         """No-op, kept for backward compatibility.

src/seriousdb/wal.py

@@ -102,6 +102,19 @@ class WriteAheadLog:
     def __init__(self, filename: str):
         self.filename = filename
         self._offset = os.path.getsize(filename) if os.path.isfile(filename) else 0
+        self._file = open(filename, "a+b")  # noqa: SIM115 - kept open, closed in close()
+
+    def close(self) -> None:
+        """Close the WAL file if it is open."""
+        f, self._file = self._file, None
+        if f is not None:
+            f.close()
+
+    def _ensure_open(self):
+        """Return the open WAL file, reopening it if necessary."""
+        if self._file is None:
+            self._file = open(self.filename, "a+b")  # noqa: SIM115 - reopened lazily, closed in close()
+        return self._file
 
     def append(self, entry: WalEntry) -> None:
         """Append `entry` to the log and fsync it.
@@ -117,12 +130,24 @@ def append(self, entry: WalEntry) -> None:
             If the log file cannot be written.
         """
         self._repair_torn_tail()
-        with open(self.filename, "ab") as f:
-            data = (json.dumps(entry.to_dict()) + "\n").encode()
+        f = self._ensure_open()
+
+        data = (json.dumps(entry.to_dict()) + "\n").encode()
+
+        try:
             f.write(data)
             f.flush()
             os.fsync(f.fileno())
-        self._offset += len(data)
+
+            self._offset += len(data)
+        except Exception:
+            try:
+                f.close()
+            except OSError:
+                pass
+            finally:
+                self._file = None
+            raise
 
     def replay(self) -> list[WalEntry]:
         """Return every entry durably written to the log, oldest first.
@@ -145,40 +170,43 @@ def replay(self) -> list[WalEntry]:
             self._offset = 0
             return []
 
+        f = self._ensure_open()
+
         entries: list[WalEntry] = []
         good_offset = 0
         found_bad_entry = False
-        with open(self.filename, "rb") as f:
-            for raw_line in f:
-                if not raw_line.endswith(b"\n"):
+
+        f.seek(0)
+
+        for raw_line in f:
+            if not raw_line.endswith(b"\n"):
+                found_bad_entry = True
+                break
+            line = raw_line.strip()
+            if line:
+                try:
+                    entries.append(WalEntry.from_dict(json.loads(line.decode())))
+                except (
+                    json.JSONDecodeError,
+                    UnicodeDecodeError,
+                    TypeError,
+                    ValueError,
+                ) as e:
+                    logger.warning(
+                        "Corrupt entry in write-ahead log %s (%s)",
+                        self.filename,
+                        e,
+                    )
                     found_bad_entry = True
                     break
-                line = raw_line.strip()
-                if line:
-                    try:
-                        entries.append(WalEntry.from_dict(json.loads(line.decode())))
-                    except (
-                        json.JSONDecodeError,
-                        UnicodeDecodeError,
-                        TypeError,
-                        ValueError,
-                    ) as e:
-                        logger.warning(
-                            "Corrupt entry in write-ahead log %s (%s)",
-                            self.filename,
-                            e,
-                        )
-                        found_bad_entry = True
-                        break
-                good_offset += len(raw_line)
+            good_offset += len(raw_line)
 
         if found_bad_entry:
             logger.warning(
                 "Truncating write-ahead log %s to its last known-good entry",
                 self.filename,
             )
-            with open(self.filename, "r+b") as f:
-                f.truncate(good_offset)
+            f.truncate(good_offset)
 
         self._offset = good_offset
         return entries
@@ -194,12 +222,19 @@ def clear(self) -> None:
         dir_name = os.path.dirname(self.filename) or "."
         with tempfile.NamedTemporaryFile("wb", dir=dir_name, delete=False) as tmp_file:
             pass
+
         try:
+            self.close()
             os.replace(tmp_file.name, self.filename)
         except OSError:
-            os.unlink(tmp_file.name)
+            try:
+                os.unlink(tmp_file.name)
+            except FileNotFoundError:
+                pass
             raise
+
         self._offset = 0
+        self._file = None
 
     def _repair_torn_tail(self) -> None:
         """Truncate any bytes left by a previous failed write.