Refactored storage layer to use per-operation auto-commit with explicit transaction API.

tests/test_commands.py

@@ -17,7 +17,7 @@
 Tests cover the @on_command decorator, CommandEvent/CommandInfo dataclasses,
 CommandRegistry (trigger mapping, alias resolution, conflict detection, module
 scanning), CommandDispatcher (parsing, permission checks, cooldowns,
-transaction commit/rollback, built-in commands), ModuleCommands
+timeout handling, built-in commands), ModuleCommands
 (ownership-scoped wrapper), and CommandContext.
 """
 
@@ -755,10 +755,10 @@ class TestCommandDispatcherDispatch:
         assert row is not None
         assert row["name"] == "apple"
 
-    async def test_exception_rolls_back(
+    async def test_exception_does_not_roll_back_committed_writes(
         self, storage: ModuleStorage, caplog: pytest.LogCaptureFixture
     ) -> None:
-        """A handler that raises has its database changes rolled back."""
+        """A handler that raises does not undo already-committed writes."""
         await storage.execute(
             "CREATE TABLE IF NOT EXISTS items (id INTEGER, name TEXT)"
         )
@@ -775,10 +775,11 @@ class TestCommandDispatcherDispatch:
             await dispatcher.dispatch(_make_chat_event(body="!add"))
 
         row = await storage.fetch_one("SELECT * FROM items WHERE id = ?", (1,))
-        assert row is None
+        assert row is not None
+        assert row["name"] == "apple"
         assert "raised exception: boom" in caplog.text
 
-    async def test_timeout_rolls_back(
+    async def test_timeout_is_logged(
         self, storage: ModuleStorage, caplog: pytest.LogCaptureFixture
     ) -> None:
         """A timed-out handler is cancelled and logged."""
@@ -796,6 +797,50 @@ class TestCommandDispatcherDispatch:
             for r in caplog.records
         )
 
+    async def test_timeout_does_not_roll_back_committed_writes(
+        self, storage: ModuleStorage
+    ) -> None:
+        """A timed-out handler's already-committed writes persist."""
+        await storage.execute(
+            "CREATE TABLE IF NOT EXISTS items (id INTEGER, name TEXT)"
+        )
+
+        async def handler(ctx: CommandContext) -> None:
+            await ctx.storage.execute(
+                "INSERT INTO items (id, name) VALUES (?, ?)", (1, "apple")
+            )
+            await asyncio.sleep(10)
+
+        dispatcher, _, _ = self._make_dispatcher(storage, handler_timeout=0.05)
+        dispatcher.register("slow", handler, module_name="mod_a")
+        await dispatcher.dispatch(_make_chat_event(body="!slow"))
+
+        row = await storage.fetch_one("SELECT * FROM items WHERE id = ?", (1,))
+        assert row is not None
+        assert row["name"] == "apple"
+
+    async def test_timeout_rolls_back_explicit_transaction(
+        self, storage: ModuleStorage
+    ) -> None:
+        """A timed-out handler's uncommitted transaction is rolled back."""
+        await storage.execute(
+            "CREATE TABLE IF NOT EXISTS items (id INTEGER, name TEXT)"
+        )
+
+        async def handler(ctx: CommandContext) -> None:
+            async with ctx.storage.transaction():
+                await ctx.storage.execute(
+                    "INSERT INTO items (id, name) VALUES (?, ?)", (1, "apple")
+                )
+                await asyncio.sleep(10)
+
+        dispatcher, _, _ = self._make_dispatcher(storage, handler_timeout=0.05)
+        dispatcher.register("slow", handler, module_name="mod_a")
+        await dispatcher.dispatch(_make_chat_event(body="!slow"))
+
+        row = await storage.fetch_one("SELECT * FROM items WHERE id = ?", (1,))
+        assert row is None
+
     async def test_builtin_command_dispatched(self, storage: ModuleStorage) -> None:
         """A built-in command receives (event, owncast_client)."""
         captured: list[tuple[Any, Any]] = []

tests/test_events.py

@@ -15,7 +15,7 @@
 """Unit tests for event registration, dispatching, and context infrastructure.
 
 Tests cover the @on_event decorator, Priority enum, EventRegistry,
-EventDispatcher (priority sorting, propagation control, timeout/rollback,
+EventDispatcher (priority sorting, propagation control, timeout handling,
 command dispatch phase), ModuleEvents (ownership checks), and
 EventContext/PropagationState.
 """
@@ -653,8 +653,8 @@ class TestEventDispatcherDispatch:
         assert call_order == ["stopper", "follower"]
 
 
-class TestEventDispatcherTransactions:
-    """Tests _call_handler() commit/rollback behaviour in EventDispatcher."""
+class TestEventDispatcherStorage:
+    """Tests _call_handler() storage behaviour in EventDispatcher."""
 
     async def test_success_commits(self, storage: ModuleStorage) -> None:
         """A successful handler's database changes are committed."""
@@ -682,16 +682,16 @@ class TestEventDispatcherTransactions:
         assert row is not None
         assert row[0] == "committed"
 
-    async def test_exception_rolls_back(
+    async def test_exception_does_not_roll_back_committed_writes(
         self, storage: ModuleStorage, caplog: pytest.LogCaptureFixture
     ) -> None:
-        """A handler that raises has its database changes rolled back."""
-        # Pre-create the table outside the handler's transaction.
+        """A handler that raises does not undo already-committed writes."""
+        # Pre-create the table outside the handler.
         await storage.execute("CREATE TABLE IF NOT EXISTS test_tbl (val TEXT)")
 
         async def handler(ctx: EventContext[Any]) -> None:
             await ctx.storage.execute(
-                "INSERT INTO test_tbl (val) VALUES (?)", ("rolled_back",)
+                "INSERT INTO test_tbl (val) VALUES (?)", ("persisted",)
             )
             raise RuntimeError("boom")
 
@@ -710,7 +710,8 @@ class TestEventDispatcherTransactions:
             await dispatcher.dispatch(EventType.CHAT, _make_chat_event())
 
         row = await storage.fetch_one("SELECT val FROM test_tbl")
-        assert row is None
+        assert row is not None
+        assert row[0] == "persisted"
         assert any("raised exception: boom" in r.message for r in caplog.records)
 
     async def test_exception_does_not_stop_next_handler(
@@ -741,10 +742,10 @@ class TestEventDispatcherTransactions:
         await dispatcher.dispatch(EventType.CHAT, _make_chat_event())
         assert call_order == ["failing", "second"]
 
-    async def test_timeout_rolls_back(
+    async def test_timeout_is_logged(
         self, storage: ModuleStorage, caplog: pytest.LogCaptureFixture
     ) -> None:
-        """A handler that exceeds the timeout is cancelled and changes rolled back."""
+        """A handler that exceeds the timeout is cancelled and logged."""
 
         async def slow_handler(ctx: EventContext[Any]) -> None:
             await asyncio.sleep(10)
@@ -767,12 +768,16 @@ class TestEventDispatcherTransactions:
             for r in caplog.records
         )
 
-    async def test_timeout_rolls_back_db_changes(self, storage: ModuleStorage) -> None:
-        """A timed-out handler's database writes are rolled back."""
+    async def test_timeout_does_not_roll_back_committed_writes(
+        self, storage: ModuleStorage
+    ) -> None:
+        """A timed-out handler's already-committed writes persist."""
         await storage.execute("CREATE TABLE IF NOT EXISTS test_tbl (val TEXT)")
 
         async def slow_handler(ctx: EventContext[Any]) -> None:
-            await ctx.storage.execute("INSERT INTO test_tbl (val) VALUES (?)", ("tmp",))
+            await ctx.storage.execute(
+                "INSERT INTO test_tbl (val) VALUES (?)", ("persisted",)
+            )
             await asyncio.sleep(10)
 
         module_ctx = _make_module_context(storage, "mod_a")
@@ -788,6 +793,36 @@ class TestEventDispatcherTransactions:
         dispatcher.register(slow_handler, (EventType.CHAT,), "mod_a")
         await dispatcher.dispatch(EventType.CHAT, _make_chat_event())
 
+        row = await storage.fetch_one("SELECT val FROM test_tbl")
+        assert row is not None
+        assert row[0] == "persisted"
+
+    async def test_timeout_rolls_back_explicit_transaction(
+        self, storage: ModuleStorage
+    ) -> None:
+        """A timed-out handler's uncommitted transaction is rolled back."""
+        await storage.execute("CREATE TABLE IF NOT EXISTS test_tbl (val TEXT)")
+
+        async def slow_handler(ctx: EventContext[Any]) -> None:
+            async with ctx.storage.transaction():
+                await ctx.storage.execute(
+                    "INSERT INTO test_tbl (val) VALUES (?)", ("uncommitted",)
+                )
+                await asyncio.sleep(10)
+
+        module_ctx = _make_module_context(storage, "mod_a")
+
+        async def command_dispatch(event: ChatEvent) -> None:
+            pass
+
+        dispatcher = EventDispatcher(
+            command_dispatch=command_dispatch,
+            get_module_context=lambda name: module_ctx,
+            handler_timeout=0.05,
+        )
+        dispatcher.register(slow_handler, (EventType.CHAT,), "mod_a")
+        await dispatcher.dispatch(EventType.CHAT, _make_chat_event())
+
         row = await storage.fetch_one("SELECT val FROM test_tbl")
         assert row is None
 

tests/test_storage.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 from typing import TYPE_CHECKING, Any
 
 import pytest
@@ -190,7 +191,7 @@ class TestFetchValue:
 
 
 class TestAutoCommit:
-    """Outside _checkout(), each operation auto-commits or auto-rolls-back."""
+    """Each operation auto-commits on success or auto-rolls-back on failure."""
 
     async def test_auto_commits_on_success(
         self, storage_with_table: ModuleStorage
@@ -226,48 +227,6 @@ class TestAutoCommit:
         assert row["name"] == "a"
 
 
-class TestCheckout:
-    """_checkout() pins a single connection for all operations."""
-
-    async def test_commit_persists(self, storage_with_table: ModuleStorage) -> None:
-        """Writes inside _checkout() become visible after _commit()."""
-        async with storage_with_table._checkout():
-            await storage_with_table.execute(
-                "INSERT INTO items (name, value) VALUES (?, ?)", ("a", 1)
-            )
-            await storage_with_table._commit()
-
-        row = await storage_with_table.fetch_one("SELECT name FROM items WHERE id = 1")
-        assert row is not None
-        assert row["name"] == "a"
-
-    async def test_rollback_discards(self, storage_with_table: ModuleStorage) -> None:
-        """Writes inside _checkout() are discarded after _rollback()."""
-        async with storage_with_table._checkout():
-            await storage_with_table.execute(
-                "INSERT INTO items (name, value) VALUES (?, ?)", ("a", 1)
-            )
-            await storage_with_table._rollback()
-
-        row = await storage_with_table.fetch_one("SELECT name FROM items WHERE id = 1")
-        assert row is None
-
-    async def test_uncommitted_writes_invisible_from_other_connection(
-        self, storage_with_table: ModuleStorage
-    ) -> None:
-        """Uncommitted writes are not visible from a separate connection."""
-        async with (
-            ModuleStorage(storage_with_table._db_path.parent, "test_module") as s2,
-            storage_with_table._checkout(),
-        ):
-            await storage_with_table.execute(
-                "INSERT INTO items (name, value) VALUES (?, ?)", ("a", 1)
-            )
-            # Before commit, s2 should not see the row.
-            row = await s2.fetch_one("SELECT name FROM items WHERE id = 1")
-            assert row is None
-
-
 class TestTransaction:
     """transaction() context manager commits on clean exit, rolls back on exception."""
 
@@ -275,7 +234,7 @@ class TestTransaction:
         self, storage_with_table: ModuleStorage
     ) -> None:
         """Writes inside transaction() persist after clean exit."""
-        async with storage_with_table._checkout(), storage_with_table.transaction():
+        async with storage_with_table.transaction():
             await storage_with_table.execute(
                 "INSERT INTO items (name, value) VALUES (?, ?)", ("a", 1)
             )
@@ -289,7 +248,7 @@ class TestTransaction:
     ) -> None:
         """Writes inside transaction() are discarded if an exception is raised."""
         with pytest.raises(RuntimeError, match="boom"):  # noqa: PT012
-            async with storage_with_table._checkout(), storage_with_table.transaction():
+            async with storage_with_table.transaction():
                 await storage_with_table.execute(
                     "INSERT INTO items (name, value) VALUES (?, ?)",
                     ("a", 1),
@@ -299,6 +258,51 @@ class TestTransaction:
         row = await storage_with_table.fetch_one("SELECT name FROM items WHERE id = 1")
         assert row is None
 
+    async def test_nested_transaction_raises(
+        self, storage_with_table: ModuleStorage
+    ) -> None:
+        """Nesting transaction() calls raises RuntimeError."""
+        async with storage_with_table.transaction():
+            with pytest.raises(RuntimeError, match="cannot be nested"):
+                async with storage_with_table.transaction():
+                    pass  # pragma: no cover
+
+    async def test_uncommitted_transaction_invisible_from_other_connection(
+        self, storage_with_table: ModuleStorage
+    ) -> None:
+        """Uncommitted writes inside a transaction are not visible externally."""
+        async with (
+            ModuleStorage(storage_with_table._db_path.parent, "test_module") as s2,
+            storage_with_table.transaction(),
+        ):
+            await storage_with_table.execute(
+                "INSERT INTO items (name, value) VALUES (?, ?)", ("a", 1)
+            )
+            # Before commit, s2 should not see the row.
+            row = await s2.fetch_one("SELECT name FROM items WHERE id = 1")
+            assert row is None
+
+    async def test_rolls_back_on_cancellation(
+        self, storage_with_table: ModuleStorage
+    ) -> None:
+        """A cancelled transaction rolls back uncommitted writes."""
+
+        async def slow_txn() -> None:
+            async with storage_with_table.transaction():
+                await storage_with_table.execute(
+                    "INSERT INTO items (name, value) VALUES (?, ?)", ("a", 1)
+                )
+                await asyncio.sleep(10)
+
+        task = asyncio.create_task(slow_txn())
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+
+        row = await storage_with_table.fetch_one("SELECT name FROM items WHERE id = 1")
+        assert row is None
+
 
 class TestConnectionPool:
     """Connection pool reuses connections and respects pool_size."""
@@ -311,21 +315,32 @@ class TestConnectionPool:
         assert len(storage._all_connections) == 1
 
     async def test_pool_grows_up_to_pool_size(self, tmp_path: Path) -> None:
-        """Concurrent checkouts grow the pool up to pool_size."""
-        async with (
-            ModuleStorage(tmp_path, "pool_test", pool_size=2) as s,
-            s._checkout(),
-            s._checkout(),
-        ):
+        """Concurrent transactions grow the pool up to pool_size."""
+        gate = asyncio.Event()
+
+        async def hold_transaction(storage: ModuleStorage) -> None:
+            async with storage.transaction():
+                gate.set()
+                await asyncio.sleep(0.5)
+
+        async with ModuleStorage(tmp_path, "pool_test", pool_size=2) as s:
+            task = asyncio.create_task(hold_transaction(s))
+            await gate.wait()
+            # First task holds one connection; acquire a second directly.
+            conn = await asyncio.wait_for(s._acquire(), timeout=1.0)
             assert len(s._all_connections) == 2
+            s._release(conn)
+            task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await task
 
     async def test_pool_exhaustion_blocks(self, tmp_path: Path) -> None:
         """When the pool is exhausted, acquire blocks until a connection is released."""
         async with (
             ModuleStorage(tmp_path, "pool_block", pool_size=1) as s,
-            s._checkout(),
+            s.transaction(),
         ):
-            # Pool is exhausted (pool_size=1, one checked out).
+            # Pool is exhausted (pool_size=1, one held by transaction).
             # A second acquire should block, so wait_for should time out.
             with pytest.raises(TimeoutError):
                 await asyncio.wait_for(s._acquire(), timeout=0.1)