Add holistic table event processing benchmark

tests/benchmarks/README.md

@@ -0,0 +1,16 @@
+# Event processing benchmarks
+
+```sh
+uv run pytest tests/benchmarks/test_event_processing.py --codspeed
+```
+
+Omit `--codspeed` for local wall-clock measurements.
+
+- `counter`: three increments 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; each batch returns to the same filter and
+sort direction. 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,9 +1,4 @@
-"""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 traceback
@@ -17,31 +12,52 @@
 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
+from .fixtures import BenchmarkState, TableState
 
 
-@pytest_asyncio.fixture
-async def event_processing_harness():
-    """Set up the full event processing pipeline for benchmarking.
+@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.
 
-    Creates a ``BaseStateEventProcessor`` wired to a real
-    ``StateManagerMemory`` with mock emit callbacks.  Events are
-    enqueued directly and deltas are collected via the emit callback.
+    Args:
+        request: Selects the counter or table workload.
 
     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]]]] = []
+    table = request.param == "table"
+    handler = (
+        TableState.event_handlers["set_status"]
+        if table
+        else BenchmarkState.event_handlers["increment"]
+    )
+    payloads = (
+        [{"status": status} for status in ("open", "", "paid") * 2]
+        if table
+        else [{}] * 3
+    )
+    events = [
+        Event(
+            name=format_event_handler(handler),
+            router_data={"query": {}, "path": "/"},
+            payload=payload,
+        )
+        for payload in payloads
+    ]
+    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
+        if table:
+            json_dumps(StateUpdate(delta=delta), separators=(",", ":"))
 
     async def emit_event_impl(token: str, *events: Event) -> None:
         pass
@@ -54,69 +70,45 @@ 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
-
-        yield run_events
-
-        await state_manager.close()
+            assert emitted == len(events)
 
+        try:
+            if table:
+                await run_events()
+            yield run_events
+        finally:
+            await state_manager.close()
 
-def test_process_event(
-    event_processing_harness,
-    benchmark: BenchmarkFixture,
-):
-    """Benchmark processing 3 increment events through the full pipeline.
 
-    The first event creates fresh state (cold path), the next two reuse
-    the existing state (warm path).  Only event processing is timed.
+def test_process_event(event_processing_harness, benchmark: BenchmarkFixture):
+    """Benchmark a batch of three 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())