Add holistic table event processing benchmark

tests/benchmarks/README.md

@@ -0,0 +1,20 @@
+# Event processing benchmarks
+
+```sh
+uv run pytest tests/benchmarks/test_event_processing.py --codspeed
+```
+
+Omit `--codspeed` for local wall-clock measurements.
+
+- `counter`: two increments and two decrements through the in-memory event
+  processor.
+- `table`: six filter/sort events over 1,000 dataclass orders, including computed
+  rows and totals and JSON encoding of each `StateUpdate`. The batch cycles
+  through open/all/paid orders twice, reversing sort direction on every event.
+
+The table is warmed before timing, and both batches return state to its
+starting point so every sample measures identical work.
+`test_table_event_deltas` separately verifies the serialized rows, sort
+direction, and totals that the table benchmark encodes.
+Timing includes processor startup/shutdown, but excludes initial
+hydration, Socket.IO packet framing, network transport, databases, and rendering.

tests/benchmarks/fixtures.py

@@ -272,6 +272,69 @@ def enter_component(
         return enter_component
 
 
+@dataclass
+class Order:
+    """An order in the table event benchmark."""
+
+    name: str
+    customer: str
+    amount: float
+    status: str
+
+
+class TableState(rx.State):
+    """A 1000-row table with filtering, sorting, and a computed total."""
+
+    orders: rx.Field[list[Order]] = rx.field(
+        default_factory=lambda: [
+            Order(
+                name=f"order {i}",
+                customer=f"customer {i % 50}",
+                amount=i * 1.5,
+                status=("open", "paid", "shipped")[i % 3],
+            )
+            for i in range(1000)
+        ]
+    )
+    status: rx.Field[str] = rx.field("")
+    sort_reverse: rx.Field[bool] = rx.field(False)
+
+    @rx.event
+    def set_status(self, status: str):
+        """Filter the table by status, flipping the sort direction.
+
+        Args:
+            status: The status to keep, or an empty string for all rows.
+        """
+        self.status = status
+        self.sort_reverse = not self.sort_reverse
+
+    @rx.var
+    def filtered_orders(self) -> list[Order]:
+        """The rows matching the filter, sorted.
+
+        Returns:
+            The filtered, sorted rows.
+        """
+        orders = self.orders
+        if self.status:
+            orders = [order for order in orders if order.status == self.status]
+        return sorted(
+            orders,
+            key=lambda order: order.amount,
+            reverse=self.sort_reverse,
+        )
+
+    @rx.var
+    def total_amount(self) -> float:
+        """The amount summed over the filtered rows.
+
+        Returns:
+            The total amount.
+        """
+        return sum(order.amount for order in self.filtered_orders)
+
+
 class BenchmarkState(rx.State):
     """State for the benchmark."""
 

tests/benchmarks/test_event_processing.py

@@ -1,47 +1,105 @@
-"""Benchmark for the event processing pipeline.
-
-Measures the time from enqueuing events via ``BaseStateEventProcessor``
-to collecting all emitted ``StateUpdate`` deltas, with mock emit
-callbacks that record the deltas.
-"""
+"""Benchmark counter and table events through the in-memory event pipeline."""
 
 import asyncio
+import json
 import traceback
-from collections.abc import Mapping
+from collections.abc import AsyncIterator, Awaitable, Callable, Mapping
+from contextlib import asynccontextmanager
 from typing import Any
 from unittest import mock
 
 import pytest
 import pytest_asyncio
 from pytest_codspeed import BenchmarkFixture
+from reflex_base.constants.state import FIELD_MARKER
 from reflex_base.event import Event
 from reflex_base.event.context import EventContext
 from reflex_base.event.processor import BaseStateEventProcessor
-from reflex_base.utils.format import format_event_handler
+from reflex_base.utils.format import format_event_handler, json_dumps
 
 from reflex.istate.manager.memory import StateManagerMemory
+from reflex.state import StateUpdate
+
+from .fixtures import BenchmarkState, TableState
+
+TABLE_STATUSES = ("open", "", "paid")
+
+
+def _encode_delta(delta: Mapping[str, Mapping[str, Any]]) -> str:
+    """Serialize a delta the way the socket path does.
+
+    Args:
+        delta: The state changes emitted by the processor.
+
+    Returns:
+        The delta encoded as a StateUpdate envelope.
+    """
+    return json_dumps(StateUpdate(delta=delta), separators=(",", ":"))
+
 
-from .fixtures import BenchmarkState
+def _events(handler_name: str, payloads: list[dict[str, Any]]) -> list[Event]:
+    """Build one event per payload for the given handler.
 
+    Args:
+        handler_name: The formatted event handler name.
+        payloads: One payload per event.
+
+    Returns:
+        The events to enqueue.
+    """
+    return [
+        Event(
+            name=handler_name,
+            router_data={"query": {}, "path": "/"},
+            payload=payload,
+        )
+        for payload in payloads
+    ]
 
-@pytest_asyncio.fixture
-async def event_processing_harness():
-    """Set up the full event processing pipeline for benchmarking.
 
-    Creates a ``BaseStateEventProcessor`` wired to a real
-    ``StateManagerMemory`` with mock emit callbacks.  Events are
-    enqueued directly and deltas are collected via the emit callback.
+def _counter_events() -> list[Event]:
+    """Two increments followed by two decrements, returning to the start.
+
+    Returns:
+        The counter event batch.
+    """
+    increment = format_event_handler(BenchmarkState.event_handlers["increment"])
+    decrement = format_event_handler(BenchmarkState.event_handlers["decrement"])
+    return _events(increment, [{}] * 2) + _events(decrement, [{}] * 2)
+
+
+def _table_events() -> list[Event]:
+    """Two filter cycles, ending on the starting sort direction.
+
+    Returns:
+        The table event batch.
+    """
+    set_status = format_event_handler(TableState.event_handlers["set_status"])
+    return _events(set_status, [{"status": status} for status in TABLE_STATUSES * 2])
+
+
+@asynccontextmanager
+async def _event_pipeline(
+    events: list[Event],
+    on_delta: Callable[[Mapping[str, Mapping[str, Any]]], Any],
+) -> AsyncIterator[Callable[[], Awaitable[None]]]:
+    """Wire a ``BaseStateEventProcessor`` to an in-memory state manager.
+
+    Args:
+        events: The batch to enqueue on each run.
+        on_delta: Called with each emitted delta.
 
     Yields:
-        An async callable that enqueues the given number of events
-        and waits for all expected deltas.
+        An async callable that processes one batch and checks its delta count.
     """
-    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]] = []
+    emitted = 0
 
     async def emit_delta_impl(  # noqa: RUF029
         token: str, delta: Mapping[str, Mapping[str, Any]]
     ) -> None:
-        emitted_deltas.append((token, delta))
+        nonlocal emitted
+        emitted += 1
+        on_delta(delta)
 
     async def emit_event_impl(token: str, *events: Event) -> None:
         pass
@@ -54,69 +112,98 @@ def handle_backend_exception(ex: Exception) -> None:
         backend_exception_handler=handle_backend_exception,
         graceful_shutdown_timeout=5,
     )
-    # Mock _rehydrate so the processor doesn't try to push full state
-    # to a non-existent frontend on the first event.
+    # Skip initial hydration because there is no frontend.
     with mock.patch.object(processor, "_rehydrate", new=mock.AsyncMock()):
         state_manager = StateManagerMemory()
-        root_context = EventContext(
+        processor._root_context = EventContext(
             token="",
             state_manager=state_manager,
             enqueue_impl=processor.enqueue_many,
             emit_delta_impl=emit_delta_impl,
             emit_event_impl=emit_event_impl,
         )
-        processor._root_context = root_context
-
-        token = "benchmark-token"
-        handler_name = format_event_handler(BenchmarkState.event_handlers["increment"])
-        event = Event(
-            name=handler_name,
-            router_data={
-                "query": {},
-                "path": "/",
-            },
-            payload={},
-        )
-
-        async def run_events(num_events: int, num_expected_deltas: int) -> None:
-            """Enqueue events and wait for all deltas to be emitted.
-
-            Args:
-                num_events: Number of increment events to enqueue.
-                num_expected_deltas: How many deltas to wait for.
-            """
-            emitted_deltas.clear()
 
+        async def run_events() -> None:
+            """Process the batch and verify that each event emitted a delta."""
+            nonlocal emitted
+            emitted = 0
             async with processor as p:
                 async for _ in asyncio.as_completed([
-                    await p.enqueue(token, event) for _ in range(num_events)
+                    await p.enqueue("benchmark-token", event) for event in events
                 ]):
                     pass
-            assert len(emitted_deltas) == num_expected_deltas
+            assert emitted == len(events)
 
-        yield run_events
+        try:
+            yield run_events
+        finally:
+            await state_manager.close()
 
-        await state_manager.close()
 
+@pytest_asyncio.fixture(params=["counter", "table"])
+async def event_processing_harness(request: pytest.FixtureRequest):
+    """Set up a fixed event batch, warming the table before timing.
 
-def test_process_event(
-    event_processing_harness,
-    benchmark: BenchmarkFixture,
-):
-    """Benchmark processing 3 increment events through the full pipeline.
+    Both batches return state to its starting point, so every benchmark
+    sample measures identical work.
 
-    The first event creates fresh state (cold path), the next two reuse
-    the existing state (warm path).  Only event processing is timed.
+    Args:
+        request: Selects the counter or table workload.
+
+    Yields:
+        An async callable that processes one batch and checks its delta count.
+    """
+    table = request.param == "table"
+    events = _table_events() if table else _counter_events()
+    on_delta = _encode_delta if table else (lambda delta: None)
+    async with _event_pipeline(events, on_delta) as run_events:
+        if table:
+            await run_events()
+        yield run_events
+
+
+def test_process_event(event_processing_harness, benchmark: BenchmarkFixture):
+    """Benchmark a batch of four counter events or six table events.
 
     Args:
-        event_processing_harness: The run_events async callable.
+        event_processing_harness: The async batch runner.
         benchmark: The codspeed benchmark fixture.
     """
-    run_events = event_processing_harness
     loop = asyncio.get_event_loop()
 
-    # Each event handler (increment) does a single state mutation with
-    # no yields, so we expect 1 delta per event = 3 total.
     @benchmark
     def _():
-        loop.run_until_complete(run_events(num_events=3, num_expected_deltas=3))
+        loop.run_until_complete(event_processing_harness())
+
+
+@pytest.mark.asyncio
+async def test_table_event_deltas():
+    """Verify filtered rows, sort direction, and totals across repeated batches."""
+    updates: list[str] = []
+
+    async with _event_pipeline(
+        _table_events(), lambda delta: updates.append(_encode_delta(delta))
+    ) as run_events:
+        for _ in range(2):
+            await run_events()
+
+    for index, update in enumerate(updates):
+        status = TABLE_STATUSES[index % len(TABLE_STATUSES)]
+        reverse = index % 2 == 0
+        expected = [
+            {
+                "name": f"order {i}",
+                "customer": f"customer {i % 50}",
+                "amount": i * 1.5,
+                "status": ("open", "paid", "shipped")[i % 3],
+            }
+            for i in sorted(range(1000), reverse=reverse)
+            if not status or ("open", "paid", "shipped")[i % 3] == status
+        ]
+        delta = json.loads(update)["delta"][TableState.get_full_name()]
+        assert delta["status" + FIELD_MARKER] == status
+        assert delta["sort_reverse" + FIELD_MARKER] == reverse
+        assert delta["filtered_orders" + FIELD_MARKER] == expected
+        assert delta["total_amount" + FIELD_MARKER] == sum(
+            row["amount"] for row in expected
+        )