Melvin Strobl
72c1d73e96
C4: a device's `viewers` (superusers by default, or members) covers the image and video messages its ports publish. Anyone else gets none of their values on the socket, in the catalogue or a flow's state, 403 for their history, for the frames they hold (the last 8 included) and for TURN credentials, no bytes when asking for them, and `bye: forbidden` for a live view. A device credential stays bounded by its dashboards. The device page sets it in its Camera section, and a tile the viewer may not see says "No access to this camera" and asks the camera for nothing. C3: - A volatile upload is held to MAX_FRAME_BYTES (4 MiB, 413 above it, read without extra copies), and so is a frame on a device's socket. - A frame the engine holds in memory is written to the ring directory before a local worker is handed it, so load_artifact finds it. - A socket the bus dropped events for catches up with a snapshot, demand included, and a device's credential is checked again per frame, cached by the flow store revision and its nonce, so a lost event never keeps an unpaired device's frames coming. - A report is at most 64 KB once its named config_schemas are put in place. - A device away for 60 s has its true bool messages set to false. - Refused frames are logged by fixed keys, a frame is named after its message, `fluksio serve` waits 5 s for open connections, a tile with a stream URL creates no demand, the 8-name cap keeps the newest names. - Every container's log rotates (json-file, 3 x 10 MB).
103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
"""What the bus costs the event loop, and that it still delivers everything."""
|
|
|
|
import asyncio
|
|
import threading
|
|
|
|
from fluksio.flow.events import 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
|