Add more info to the logs and error messages

Author: daniil-quix

quixstreams/state/rocksdb/partition.py

@@ -355,6 +355,7 @@ def _open_rocksdict(self) -> Rdict:
             options=options,
             access_type=AccessType.read_write(),
         )
+        # TODO: Add docs
 
         try:
             rdict = create_rdict()
@@ -363,18 +364,22 @@ def _open_rocksdict(self) -> Rdict:
                 raise
             elif not self._changelog_producer:
                 raise RocksDBCorruptedError(
-                    "State is corrupted and cannot be recovered from changelog: "
+                    f'State store at "{self._path}" is corrupted '
+                    f"and cannot be recovered from the changelog topic: "
                     "`use_changelog_topics` is set to False."
-                )
+                ) from exc
             elif not self._options.on_corrupted_recreate:
                 raise RocksDBCorruptedError(
-                    "State is corrupted but may be recovered from changelog. "
-                    "Set `RocksDBOptions(..., on_corrupted_recreate=True)` "
-                    "to destroy the state and recreate it from changelog."
-                )
-
-            logger.info("Recreating corrupted RocksDB...")
+                    f'State store at "{self._path}" is corrupted '
+                    f"but may be recovered from the changelog topic. "
+                    "Pass `rocksdb_options=RocksDBOptions(..., on_corrupted_recreate=True)` "
+                    "to the Application to destroy the corrupted state "
+                    "and recover it from the changelog."
+                ) from exc
+
+            logger.warning(f"Destroying corrupted RocksDB path={self._path}")
             Rdict.destroy(self._path)
+            logger.warning(f"Recreating corrupted RocksDB path={self._path}")
             rdict = create_rdict()
 
         # Ensure metadata column family is created without defining it upfront

tests/test_quixstreams/test_state/test_rocksdb/test_rocksdb_partition.py

@@ -1,4 +1,3 @@
-import re
 import time
 from pathlib import Path
 from unittest.mock import patch
@@ -73,32 +72,29 @@ def test_db_corrupted_fails_with_no_changelog(
         self, store_partition_factory, tmp_path
     ):
         # Initialize and corrupt the database by messing with the MANIFEST
-        Rdict(path=tmp_path.as_posix())
+        path = tmp_path.as_posix()
+        Rdict(path=path)
         next(tmp_path.glob("MANIFEST*")).write_bytes(b"")
 
         with pytest.raises(
             RocksDBCorruptedError,
-            match=(
-                "State is corrupted and cannot be recovered from changelog: "
-                "`use_changelog_topics` is set to False."
-            ),
+            match=f'State store at "{path}" is corrupted and cannot be recovered '
+            f"from the changelog topic",
         ):
             store_partition_factory(changelog_producer=None)
 
     def test_db_corrupted_fails_with_on_corrupted_recreate_false(
         self, store_partition_factory, tmp_path
     ):
         # Initialize and corrupt the database by messing with the MANIFEST
-        Rdict(path=tmp_path.as_posix())
+        path = tmp_path.as_posix()
+        Rdict(path=path)
         next(tmp_path.glob("MANIFEST*")).write_bytes(b"")
 
         with pytest.raises(
             RocksDBCorruptedError,
-            match=re.escape(
-                "State is corrupted but may be recovered from changelog. "
-                "Set `RocksDBOptions(..., on_corrupted_recreate=True)` "
-                "to destroy the state and recreate it from changelog."
-            ),
+            match=f'State store at "{path}" is corrupted but may be recovered '
+            f"from the changelog topic",
         ):
             store_partition_factory()