From 6f33d447d29e93cb59316a9da32f33fec057b7a7 Mon Sep 17 00:00:00 2001 From: sahil-singh1107 Date: Mon, 28 Sep 2026 12:45:47 +0530 Subject: [PATCH 01/10] fix(wal): improve replay and recovery --- src/seriousdb/wal.py | 41 +++++++++++++++++++++++++++++++---------- 1 file changed, 31 insertions(+), 10 deletions(-) diff --git a/src/seriousdb/wal.py b/src/seriousdb/wal.py index 99684b3..84cec01 100644 --- a/src/seriousdb/wal.py +++ b/src/seriousdb/wal.py @@ -115,6 +115,10 @@ 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") + + def close(self) -> None: + self._file.close() def append(self, entry: WalEntry) -> None: """Append `entry` to the log and fsync it. @@ -130,11 +134,13 @@ 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.write(data) - f.flush() - os.fsync(f.fileno()) + + data = (json.dumps(entry.to_dict()) + "\n").encode() + + self._file.write(data) + self._file.flush() + os.fsync(self._file.fileno()) + self._offset += len(data) def replay(self) -> list[WalEntry]: @@ -161,9 +167,23 @@ def replay(self) -> list[WalEntry]: 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"): + + self._file.seek(0) + + for raw_line in self._file: + 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, 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() @@ -190,8 +210,7 @@ def replay(self) -> list[WalEntry]: "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) + self._file.truncate(good_offset) self._offset = good_offset return entries @@ -208,7 +227,9 @@ def clear(self) -> None: with tempfile.NamedTemporaryFile("wb", dir=dir_name, delete=False) as tmp_file: pass try: + self._file.close() os.replace(tmp_file.name, self.filename) + self._file = open(self.filename, "a+b") except OSError: os.unlink(tmp_file.name) raise From c9e473c42ccc6e21e6666e9f8865a81e80464df2 Mon Sep 17 00:00:00 2001 From: sahil-singh1107 Date: Wed, 30 Sep 2026 21:40:58 +0530 Subject: [PATCH 02/10] fix: manage WAL file handles during load, clear and unsuccessful writes --- src/seriousdb/cache.py | 12 +++++++----- src/seriousdb/wal.py | 42 +++++++++++++++++++++++++++++++++--------- 2 files changed, 40 insertions(+), 14 deletions(-) diff --git a/src/seriousdb/cache.py b/src/seriousdb/cache.py index 301a1a7..6b84abd 100644 --- a/src/seriousdb/cache.py +++ b/src/seriousdb/cache.py @@ -321,11 +321,13 @@ def load(self, filename: str) -> None: e, backup, ) - db = _write_default(filename) - - wal = WriteAheadLog(f"{filename}.wal") - replayed = wal.replay() - + self.db = _write_default(filename) + self.filename = filename + if self.wal is not None: + self.wal.close() + self.wal = WriteAheadLog(f"{filename}.wal") + replayed = self.wal.replay() + self._writes_since_compact = len(replayed) for entry in replayed: entry.apply(db) diff --git a/src/seriousdb/wal.py b/src/seriousdb/wal.py index 84cec01..df54691 100644 --- a/src/seriousdb/wal.py +++ b/src/seriousdb/wal.py @@ -118,7 +118,13 @@ def __init__(self, filename: str): self._file = open(filename, "a+b") def close(self) -> None: - self._file.close() + if self._file is not None: + self._file.close() + self._file = None + + def _ensure_open(self): + if self._file is None: + self._file = open(self.filename, "a+b") def append(self, entry: WalEntry) -> None: """Append `entry` to the log and fsync it. @@ -134,14 +140,24 @@ def append(self, entry: WalEntry) -> None: If the log file cannot be written. """ self._repair_torn_tail() + self._ensure_open() data = (json.dumps(entry.to_dict()) + "\n").encode() - self._file.write(data) - self._file.flush() - os.fsync(self._file.fileno()) - - self._offset += len(data) + try: + self._file.write(data) + self._file.flush() + os.fsync(self._file.fileno()) + + self._offset += len(data) + except Exception: + try: + self._file.close() + except OSError: + pass + finally: + self._file = None + raise def replay(self) -> list[WalEntry]: """Return every entry durably written to the log, oldest first. @@ -164,6 +180,8 @@ def replay(self) -> list[WalEntry]: self._offset = 0 return [] + self._ensure_open() + entries: list[WalEntry] = [] good_offset = 0 found_bad_entry = False @@ -226,14 +244,20 @@ 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 + + self.close() + try: - self._file.close() os.replace(tmp_file.name, self.filename) - self._file = open(self.filename, "a+b") 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. From 86f376e8c16dbb99275fdf51af74f06d3c0edae0 Mon Sep 17 00:00:00 2001 From: sahil-singh1107 Date: Wed, 30 Sep 2026 21:56:45 +0530 Subject: [PATCH 03/10] add: simple benchmark test for wal append --- benchmarks/test_wal.py | 46 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 benchmarks/test_wal.py diff --git a/benchmarks/test_wal.py b/benchmarks/test_wal.py new file mode 100644 index 0000000..a6bb8ff --- /dev/null +++ b/benchmarks/test_wal.py @@ -0,0 +1,46 @@ +"""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, + ) From 6c607004fd89358990a27baaa5e5733aa37981c3 Mon Sep 17 00:00:00 2001 From: sahil-singh1107 Date: Thu, 8 Oct 2026 15:16:26 +0530 Subject: [PATCH 04/10] resolve merge conflicts --- src/seriousdb/wal.py | 25 ++++--------------------- 1 file changed, 4 insertions(+), 21 deletions(-) diff --git a/src/seriousdb/wal.py b/src/seriousdb/wal.py index df54691..28124c6 100644 --- a/src/seriousdb/wal.py +++ b/src/seriousdb/wal.py @@ -117,7 +117,7 @@ def __init__(self, filename: str): self._offset = os.path.getsize(filename) if os.path.isfile(filename) else 0 self._file = open(filename, "a+b") - def close(self) -> None: + def _close(self) -> None: if self._file is not None: self._file.close() self._file = None @@ -196,7 +196,7 @@ def replay(self) -> list[WalEntry]: if line: try: entries.append(WalEntry.from_dict(json.loads(line.decode()))) - except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as e: + except (json.JSONDecodeError, UnicodeDecodeError, TypeError, ValueError) as e: logger.warning( "Corrupt entry in write-ahead log %s (%s)", self.filename, @@ -204,24 +204,7 @@ def replay(self) -> list[WalEntry]: ) 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( @@ -245,7 +228,7 @@ def clear(self) -> None: with tempfile.NamedTemporaryFile("wb", dir=dir_name, delete=False) as tmp_file: pass - self.close() + self._close() try: os.replace(tmp_file.name, self.filename) From 18700b530c2643ff020d03b4c7cd573ef93cc631 Mon Sep 17 00:00:00 2001 From: sahil-singh1107 Date: Mon, 12 Oct 2026 18:58:35 +0530 Subject: [PATCH 05/10] fix Cache.load --- src/seriousdb/cache.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/seriousdb/cache.py b/src/seriousdb/cache.py index 6b84abd..a1013d0 100644 --- a/src/seriousdb/cache.py +++ b/src/seriousdb/cache.py @@ -321,19 +321,16 @@ def load(self, filename: str) -> None: e, backup, ) - self.db = _write_default(filename) - self.filename = filename + db = _write_default(filename) if self.wal is not None: self.wal.close() self.wal = WriteAheadLog(f"{filename}.wal") replayed = self.wal.replay() - self._writes_since_compact = len(replayed) for entry in replayed: entry.apply(db) self.db = db self.filename = filename - self.wal = wal self._writes_since_compact = len(replayed) def flush(self) -> None: From eeed73fcb80c8d1f006d17f2f1f071952962d736 Mon Sep 17 00:00:00 2001 From: chrollo-lucifer-12 Date: Mon, 12 Oct 2026 13:38:45 +0000 Subject: [PATCH 06/10] fix: keep cache uninitialized when WAL replay fails --- src/seriousdb/cache.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/src/seriousdb/cache.py b/src/seriousdb/cache.py index a1013d0..78452ac 100644 --- a/src/seriousdb/cache.py +++ b/src/seriousdb/cache.py @@ -322,16 +322,25 @@ def load(self, filename: str) -> None: backup, ) db = _write_default(filename) - if self.wal is not None: - self.wal.close() - self.wal = WriteAheadLog(f"{filename}.wal") - replayed = self.wal.replay() - for entry in replayed: - entry.apply(db) + wal = WriteAheadLog(f"{filename}.wal") + 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. From 862f1cd8b1890bca6f3236891c2b1b6441e63ceb Mon Sep 17 00:00:00 2001 From: chrollo-lucifer-12 Date: Mon, 12 Oct 2026 13:43:41 +0000 Subject: [PATCH 07/10] fix: add docstring for close and _ensure_open --- src/seriousdb/wal.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/seriousdb/wal.py b/src/seriousdb/wal.py index 28124c6..027f620 100644 --- a/src/seriousdb/wal.py +++ b/src/seriousdb/wal.py @@ -117,12 +117,14 @@ def __init__(self, filename: str): self._offset = os.path.getsize(filename) if os.path.isfile(filename) else 0 self._file = open(filename, "a+b") - def _close(self) -> None: - if self._file is not None: - self._file.close() - self._file = None + 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") From 0db2a0c816ebfc11f388f3a3fb8c95c7096282f2 Mon Sep 17 00:00:00 2001 From: chrollo-lucifer-12 Date: Thu, 15 Oct 2026 16:21:37 +0000 Subject: [PATCH 08/10] fix the format checks and benchmark test --- benchmarks/test_wal.py | 38 ++++++++++++++++++++++++++++++++++++++ src/seriousdb/wal.py | 22 ++++++++++++++++------ 2 files changed, 54 insertions(+), 6 deletions(-) diff --git a/benchmarks/test_wal.py b/benchmarks/test_wal.py index a6bb8ff..9742738 100644 --- a/benchmarks/test_wal.py +++ b/benchmarks/test_wal.py @@ -44,3 +44,41 @@ def close_wal(wal: WriteAheadLog) -> None: 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, + ) diff --git a/src/seriousdb/wal.py b/src/seriousdb/wal.py index 027f620..108c980 100644 --- a/src/seriousdb/wal.py +++ b/src/seriousdb/wal.py @@ -115,18 +115,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") + 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: + """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") + 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. @@ -198,7 +199,12 @@ def replay(self) -> list[WalEntry]: if line: try: entries.append(WalEntry.from_dict(json.loads(line.decode()))) - except (json.JSONDecodeError, UnicodeDecodeError, TypeError, ValueError) as e: + except ( + json.JSONDecodeError, + UnicodeDecodeError, + TypeError, + ValueError, + ) as e: logger.warning( "Corrupt entry in write-ahead log %s (%s)", self.filename, @@ -230,9 +236,13 @@ def clear(self) -> None: with tempfile.NamedTemporaryFile("wb", dir=dir_name, delete=False) as tmp_file: pass +<<<<<<< HEAD self._close() +======= +>>>>>>> e583d10 (fix the format checks and benchmark test) try: + self.close() os.replace(tmp_file.name, self.filename) except OSError: try: From 7484b46d80ea0b8234fa545fc96a90dcc6f17574 Mon Sep 17 00:00:00 2001 From: chrollo-lucifer-12 Date: Fri, 30 Oct 2026 16:27:44 +0000 Subject: [PATCH 09/10] fix(wal): use close when clearing log --- src/seriousdb/wal.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/seriousdb/wal.py b/src/seriousdb/wal.py index 108c980..41deee7 100644 --- a/src/seriousdb/wal.py +++ b/src/seriousdb/wal.py @@ -236,11 +236,6 @@ def clear(self) -> None: with tempfile.NamedTemporaryFile("wb", dir=dir_name, delete=False) as tmp_file: pass -<<<<<<< HEAD - self._close() - -======= ->>>>>>> e583d10 (fix the format checks and benchmark test) try: self.close() os.replace(tmp_file.name, self.filename) From 4a76d3f0f820257710e3321a0948bfd6b4cdb5dd Mon Sep 17 00:00:00 2001 From: chrollo-lucifer-12 Date: Fri, 30 Oct 2026 16:52:57 +0000 Subject: [PATCH 10/10] perf: add configurable sync_every batching and benchmarks --- benchmarks/test_wal.py | 21 +++++++++++++-------- src/seriousdb/wal.py | 13 +++++++++---- 2 files changed, 22 insertions(+), 12 deletions(-) diff --git a/benchmarks/test_wal.py b/benchmarks/test_wal.py index 9742738..e7ddea5 100644 --- a/benchmarks/test_wal.py +++ b/benchmarks/test_wal.py @@ -21,7 +21,7 @@ def test_wal_append( def prepare_wal(): database_file.unlink(missing_ok=True) - wal = WriteAheadLog(str(database_file)) + wal = WriteAheadLog(str(database_file), sync_every=1) return (wal,), {} def append_entry(wal: WriteAheadLog) -> None: @@ -33,6 +33,7 @@ def close_wal(wal: WriteAheadLog) -> None: benchmark.extra_info.update( operation="WAL append", persistence="write + flush + fsync", + sync_every=1, key_bytes=len(entry.key.encode()), value_bytes=len(entry.value.encode()), ) @@ -46,31 +47,35 @@ def close_wal(wal: WriteAheadLog) -> None: ) -@pytest.mark.benchmark(group="wal") +@pytest.mark.benchmark(group="wal-append-many") +@pytest.mark.parametrize("sync_every", [1, 10, 50, 100, 500, 1000]) def test_wal_append_many( benchmark, database_file: Path, + sync_every: int, ) -> None: - """Measure repeated appends using one open WAL.""" + """Measure repeated appends using one open WAL across different sync_every intervals.""" entry = SetEntry(key="key", value="value") + entries_count = 1000 def prepare_wal(): database_file.unlink(missing_ok=True) - wal = WriteAheadLog(str(database_file)) + wal = WriteAheadLog(str(database_file), sync_every=sync_every) return (wal,), {} def append_many(wal: WriteAheadLog) -> None: - for _ in range(1000): + for _ in range(entries_count): 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, + operation=f"{entries_count} WAL appends (sync_every={sync_every})", + persistence=f"write + flush + fsync every {sync_every}", + sync_every=sync_every, + entries=entries_count, key_bytes=len(entry.key.encode()), value_bytes=len(entry.value.encode()), ) diff --git a/src/seriousdb/wal.py b/src/seriousdb/wal.py index 41deee7..012b894 100644 --- a/src/seriousdb/wal.py +++ b/src/seriousdb/wal.py @@ -112,8 +112,10 @@ class WriteAheadLog: Path of the log file. """ - def __init__(self, filename: str): + def __init__(self, filename: str, sync_every: int = 1): self.filename = filename + self.sync_every = sync_every + self._unsynced_count = 0 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() @@ -149,10 +151,13 @@ def append(self, entry: WalEntry) -> None: try: self._file.write(data) - self._file.flush() - os.fsync(self._file.fileno()) - self._offset += len(data) + self._unsynced_count += 1 + + if self._unsynced_count >= self.sync_every: + self._file.flush() + os.fsync(self._file.fileno()) + self._unsynced_count = 0 except Exception: try: self._file.close()