stroblme
75b3a93094
Docs / docs (push) Successful in 33s
Playwright Tests / test-playwright (1, 2) (push) Failing after 1m37s
Playwright Tests / test-playwright (2, 2) (push) Failing after 27s
pre-commit / pre-commit (push) Failing after 2m37s
Test Backend / test-backend (3.12) (push) Successful in 5m7s
Test Backend / test-backend (3.14) (push) Successful in 5m2s
Compose Smoke Test / test-compose (push) Failing after 28s
Playwright Tests / merge-reports (push) Failing after 2m29s
194 lines
6.9 KiB
Python
194 lines
6.9 KiB
Python
"""What the bus costs the event loop, and that it still delivers everything."""
|
|
|
|
import asyncio
|
|
import threading
|
|
from contextlib import AsyncExitStack
|
|
|
|
import pytest
|
|
|
|
from fluksio.flow.events import DISPATCH_BUDGET, QUEUE_SIZE, EventBus
|
|
|
|
|
|
def _event(i: int) -> dict[str, object]:
|
|
return {"type": "node_executed", "node": f"n{i}", "outputs": 1}
|
|
|
|
|
|
def test_a_cascade_costs_one_loop_callback() -> None:
|
|
"""The whole point of coalescing.
|
|
|
|
A three-node cascade publishes 13-16 events and each used to cross to the
|
|
event loop on its own. They are one callback now — and still every event.
|
|
"""
|
|
|
|
async def run() -> tuple[int, int]:
|
|
bus = EventBus()
|
|
loop = asyncio.get_running_loop()
|
|
bus.bind(loop)
|
|
calls = 0
|
|
real = loop.call_soon_threadsafe
|
|
|
|
def counting(*args: object, **kwargs: object) -> object:
|
|
nonlocal calls
|
|
calls += 1
|
|
return real(*args, **kwargs) # type: ignore[arg-type]
|
|
|
|
loop.call_soon_threadsafe = counting # type: ignore[method-assign]
|
|
try:
|
|
async with bus.subscribe() as queue:
|
|
# From a worker thread, which is where nodes run.
|
|
thread = threading.Thread(
|
|
target=lambda: [bus.publish(_event(i)) for i in range(16)]
|
|
)
|
|
thread.start()
|
|
thread.join()
|
|
await asyncio.sleep(0.05)
|
|
return calls, queue.qsize()
|
|
finally:
|
|
loop.call_soon_threadsafe = real # type: ignore[method-assign]
|
|
|
|
calls, delivered = asyncio.run(run())
|
|
assert calls == 1
|
|
assert delivered == 16
|
|
|
|
|
|
def test_a_slow_subscriber_still_loses_its_oldest() -> None:
|
|
"""Coalescing must not turn the drop-oldest bound into unbounded growth."""
|
|
|
|
async def run() -> tuple[int, object]:
|
|
bus = EventBus()
|
|
bus.bind(asyncio.get_running_loop())
|
|
async with bus.subscribe() as queue:
|
|
thread = threading.Thread(
|
|
target=lambda: [bus.publish(_event(i)) for i in range(QUEUE_SIZE + 50)]
|
|
)
|
|
thread.start()
|
|
thread.join()
|
|
await asyncio.sleep(0.05)
|
|
return queue.qsize(), (await queue.get())["node"]
|
|
|
|
size, oldest = asyncio.run(run())
|
|
assert size == QUEUE_SIZE
|
|
# The first fifty were dropped, not the last.
|
|
assert oldest == "n50"
|
|
|
|
|
|
def test_a_subscriber_that_lost_events_is_told_once() -> None:
|
|
"""What a device's socket catches up by: it may have lost its demand."""
|
|
|
|
async def run() -> tuple[bool, bool, bool]:
|
|
bus = EventBus()
|
|
bus.bind(asyncio.get_running_loop())
|
|
async with bus.subscribe() as kept, bus.subscribe() as behind:
|
|
for i in range(QUEUE_SIZE):
|
|
bus.publish(_event(i))
|
|
await asyncio.sleep(0.01)
|
|
while not kept.empty():
|
|
kept.get_nowait()
|
|
bus.publish(_event(QUEUE_SIZE))
|
|
await asyncio.sleep(0.01)
|
|
return bus.dropped(kept), bus.dropped(behind), bus.dropped(behind)
|
|
|
|
assert asyncio.run(run()) == (False, True, False)
|
|
|
|
|
|
def test_nothing_is_buffered_with_nobody_listening() -> None:
|
|
"""An engine with no browser open holds no events."""
|
|
|
|
async def run() -> int:
|
|
bus = EventBus()
|
|
bus.bind(asyncio.get_running_loop())
|
|
for i in range(10):
|
|
bus.publish(_event(i))
|
|
await asyncio.sleep(0.01)
|
|
return len(bus._pending)
|
|
|
|
assert asyncio.run(run()) == 0
|
|
|
|
|
|
@pytest.mark.parametrize("flush", [False, True])
|
|
def test_overflow_keeps_errors_and_control_events_in_order(flush: bool) -> None:
|
|
"""The staging bound and a slow subscriber use the same eviction policy."""
|
|
|
|
async def run() -> None:
|
|
bus = EventBus()
|
|
bus.bind(asyncio.get_running_loop())
|
|
critical = [
|
|
{"type": "node_log", "level": "error"},
|
|
{"type": "node_error"},
|
|
{"type": "run_started"},
|
|
{"type": "audit"},
|
|
{"type": "dashboard_changed"},
|
|
{"type": "future_control_event"},
|
|
]
|
|
async with bus.subscribe() as first, bus.subscribe() as second:
|
|
sent = [*critical, *(_event(i) for i in range(QUEUE_SIZE + 50))]
|
|
sent.append({"type": "run_finished"})
|
|
for sequence, event in enumerate(sent):
|
|
event["sequence"] = sequence
|
|
bus.publish(event)
|
|
assert len(bus._pending) <= QUEUE_SIZE
|
|
if flush:
|
|
# Queue overflow, with no staging overflow.
|
|
await asyncio.sleep(0)
|
|
await asyncio.sleep(0)
|
|
for queue in (first, second):
|
|
received = [queue.get_nowait() for _ in range(queue.qsize())]
|
|
assert len(received) == QUEUE_SIZE
|
|
assert received[: len(critical)] == critical
|
|
assert received[-1]["type"] == "run_finished"
|
|
assert received == sorted(received, key=lambda e: e["sequence"])
|
|
assert bus.dropped(queue)
|
|
assert not bus.dropped(queue)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_a_full_critical_backlog_rejects_telemetry_and_reports_loss() -> None:
|
|
async def run() -> None:
|
|
bus = EventBus()
|
|
bus.bind(asyncio.get_running_loop())
|
|
async with bus.subscribe() as queue:
|
|
for i in range(QUEUE_SIZE + 1):
|
|
bus.publish({"type": "node_error", "sequence": i})
|
|
bus.publish(_event(0))
|
|
assert len(bus._pending) == QUEUE_SIZE
|
|
await asyncio.sleep(0)
|
|
assert bus.dropped(queue)
|
|
assert (await queue.get())["sequence"] == 1
|
|
bus.publish({"type": "audit"})
|
|
await asyncio.sleep(0)
|
|
bus.publish(_event(1))
|
|
await asyncio.sleep(0)
|
|
assert bus.dropped(queue)
|
|
assert queue.qsize() == QUEUE_SIZE
|
|
assert all(
|
|
queue.get_nowait()["type"] != "node_executed" for _ in range(QUEUE_SIZE)
|
|
)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_fanout_yields_before_all_subscribers_are_served() -> None:
|
|
async def run() -> None:
|
|
bus = EventBus()
|
|
loop = asyncio.get_running_loop()
|
|
bus.bind(loop)
|
|
async with AsyncExitStack() as stack:
|
|
queues = [
|
|
await stack.enter_async_context(bus.subscribe())
|
|
for _ in range(DISPATCH_BUDGET // QUEUE_SIZE + 1)
|
|
]
|
|
for i in range(QUEUE_SIZE):
|
|
bus.publish(_event(i))
|
|
heartbeat: asyncio.Future[int] = loop.create_future()
|
|
loop.call_soon(
|
|
lambda: heartbeat.set_result(sum(queue.qsize() for queue in queues))
|
|
)
|
|
assert await heartbeat == DISPATCH_BUDGET
|
|
await asyncio.sleep(0)
|
|
assert all(queue.qsize() == QUEUE_SIZE for queue in queues)
|
|
assert not any(bus.dropped(queue) for queue in queues)
|
|
assert not bus._flush_scheduled
|
|
|
|
asyncio.run(run())
|