server: debounce item state saves and add schema contract tests
This commit is contained in:
@@ -73,6 +73,8 @@ CLIENT_PACKET_ADAPTER = TypeAdapter(ClientPacket)
|
|||||||
MAX_ACTIVE_PIANO_KEYS_PER_CLIENT = 12
|
MAX_ACTIVE_PIANO_KEYS_PER_CLIENT = 12
|
||||||
PIANO_RECORDING_MAX_MS = 30_000
|
PIANO_RECORDING_MAX_MS = 30_000
|
||||||
PIANO_RECORDING_MAX_EVENTS = 4096
|
PIANO_RECORDING_MAX_EVENTS = 4096
|
||||||
|
STATE_SAVE_DEBOUNCE_MS = 200
|
||||||
|
STATE_SAVE_MAX_DELAY_MS = 1000
|
||||||
|
|
||||||
|
|
||||||
class SignalingServer:
|
class SignalingServer:
|
||||||
@@ -103,6 +105,8 @@ class SignalingServer:
|
|||||||
self.grid_size = max(1, grid_size)
|
self.grid_size = max(1, grid_size)
|
||||||
self.instance_id = str(uuid.uuid4())
|
self.instance_id = str(uuid.uuid4())
|
||||||
self.server_version = self._resolve_server_version()
|
self.server_version = self._resolve_server_version()
|
||||||
|
self._pending_state_save_handle: asyncio.TimerHandle | None = None
|
||||||
|
self._pending_state_save_started_at: float | None = None
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _resolve_server_version() -> str:
|
def _resolve_server_version() -> str:
|
||||||
@@ -139,6 +143,32 @@ class SignalingServer:
|
|||||||
|
|
||||||
return nickname.casefold()
|
return nickname.casefold()
|
||||||
|
|
||||||
|
def _flush_state_save(self) -> None:
|
||||||
|
"""Immediately flush pending state persistence and clear debounce state."""
|
||||||
|
|
||||||
|
if self._pending_state_save_handle is not None:
|
||||||
|
self._pending_state_save_handle.cancel()
|
||||||
|
self._pending_state_save_handle = None
|
||||||
|
self._pending_state_save_started_at = None
|
||||||
|
self.item_service.save_state()
|
||||||
|
|
||||||
|
def _request_state_save(self) -> None:
|
||||||
|
"""Debounce/coalesce item-state persistence to reduce write churn."""
|
||||||
|
|
||||||
|
loop = asyncio.get_running_loop()
|
||||||
|
now = loop.time()
|
||||||
|
if self._pending_state_save_started_at is None:
|
||||||
|
self._pending_state_save_started_at = now
|
||||||
|
elapsed_ms = int((now - self._pending_state_save_started_at) * 1000)
|
||||||
|
if elapsed_ms >= STATE_SAVE_MAX_DELAY_MS:
|
||||||
|
self._flush_state_save()
|
||||||
|
return
|
||||||
|
if self._pending_state_save_handle is not None:
|
||||||
|
self._pending_state_save_handle.cancel()
|
||||||
|
remaining_ms = max(0, STATE_SAVE_MAX_DELAY_MS - elapsed_ms)
|
||||||
|
delay_ms = min(STATE_SAVE_DEBOUNCE_MS, remaining_ms)
|
||||||
|
self._pending_state_save_handle = loop.call_later(delay_ms / 1000, self._flush_state_save)
|
||||||
|
|
||||||
def _is_nickname_taken(self, nickname: str, exclude_client_id: str | None = None) -> bool:
|
def _is_nickname_taken(self, nickname: str, exclude_client_id: str | None = None) -> bool:
|
||||||
"""Check whether nickname is already used by another active client."""
|
"""Check whether nickname is already used by another active client."""
|
||||||
|
|
||||||
@@ -372,7 +402,7 @@ class SignalingServer:
|
|||||||
item.params.pop("recordingLengthMs", None)
|
item.params.pop("recordingLengthMs", None)
|
||||||
item.updatedAt = self.item_service.now_ms()
|
item.updatedAt = self.item_service.now_ms()
|
||||||
item.version += 1
|
item.version += 1
|
||||||
self.item_service.save_state()
|
self._request_state_save()
|
||||||
await self._broadcast_item(item)
|
await self._broadcast_item(item)
|
||||||
owner_id = str(session.get("ownerClientId", ""))
|
owner_id = str(session.get("ownerClientId", ""))
|
||||||
owner = self._get_client_by_id(owner_id) if owner_id else None
|
owner = self._get_client_by_id(owner_id) if owner_id else None
|
||||||
@@ -601,14 +631,17 @@ class SignalingServer:
|
|||||||
|
|
||||||
protocol = "wss" if self._ssl_context else "ws"
|
protocol = "wss" if self._ssl_context else "ws"
|
||||||
LOGGER.info("starting signaling server on %s://%s:%d", protocol, self.host, self.port)
|
LOGGER.info("starting signaling server on %s://%s:%d", protocol, self.host, self.port)
|
||||||
async with serve(
|
try:
|
||||||
self._handle_client,
|
async with serve(
|
||||||
self.host,
|
self._handle_client,
|
||||||
self.port,
|
self.host,
|
||||||
ssl=self._ssl_context,
|
self.port,
|
||||||
max_size=self.max_message_size,
|
ssl=self._ssl_context,
|
||||||
):
|
max_size=self.max_message_size,
|
||||||
await asyncio.Future()
|
):
|
||||||
|
await asyncio.Future()
|
||||||
|
finally:
|
||||||
|
self._flush_state_save()
|
||||||
|
|
||||||
async def _handle_client(self, websocket: ServerConnection) -> None:
|
async def _handle_client(self, websocket: ServerConnection) -> None:
|
||||||
"""Handle one websocket client's connect/message/disconnect lifecycle."""
|
"""Handle one websocket client's connect/message/disconnect lifecycle."""
|
||||||
@@ -631,7 +664,7 @@ class SignalingServer:
|
|||||||
await self._finalize_piano_recording(item_id)
|
await self._finalize_piano_recording(item_id)
|
||||||
for item in self.item_service.drop_carried_items_for_disconnect(disconnected):
|
for item in self.item_service.drop_carried_items_for_disconnect(disconnected):
|
||||||
await self._broadcast_item(item)
|
await self._broadcast_item(item)
|
||||||
self.item_service.save_state()
|
self._request_state_save()
|
||||||
LOGGER.info(
|
LOGGER.info(
|
||||||
"client disconnected id=%s nickname=%s total=%d",
|
"client disconnected id=%s nickname=%s total=%d",
|
||||||
disconnected.id,
|
disconnected.id,
|
||||||
@@ -865,7 +898,7 @@ class SignalingServer:
|
|||||||
item = self.item_service.default_item(client, packet.itemType)
|
item = self.item_service.default_item(client, packet.itemType)
|
||||||
self.item_service.add_item(item)
|
self.item_service.add_item(item)
|
||||||
await self._broadcast_item(item)
|
await self._broadcast_item(item)
|
||||||
self.item_service.save_state()
|
self._request_state_save()
|
||||||
LOGGER.info(
|
LOGGER.info(
|
||||||
"item created by=%s item_id=%s type=%s title=%s x=%d y=%d",
|
"item created by=%s item_id=%s type=%s title=%s x=%d y=%d",
|
||||||
client.nickname,
|
client.nickname,
|
||||||
@@ -913,7 +946,7 @@ class SignalingServer:
|
|||||||
item.y = client.y
|
item.y = client.y
|
||||||
item.updatedAt = self.item_service.now_ms()
|
item.updatedAt = self.item_service.now_ms()
|
||||||
await self._broadcast_item(item)
|
await self._broadcast_item(item)
|
||||||
self.item_service.save_state()
|
self._request_state_save()
|
||||||
await self._send_item_result(client, True, "pickup", f"Picked up {item.title}.", item.id)
|
await self._send_item_result(client, True, "pickup", f"Picked up {item.title}.", item.id)
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -933,7 +966,7 @@ class SignalingServer:
|
|||||||
item.y = packet.y
|
item.y = packet.y
|
||||||
item.updatedAt = self.item_service.now_ms()
|
item.updatedAt = self.item_service.now_ms()
|
||||||
await self._broadcast_item(item)
|
await self._broadcast_item(item)
|
||||||
self.item_service.save_state()
|
self._request_state_save()
|
||||||
await self._send_item_result(client, True, "drop", f"Dropped {item.title}.", item.id)
|
await self._send_item_result(client, True, "drop", f"Dropped {item.title}.", item.id)
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -968,7 +1001,7 @@ class SignalingServer:
|
|||||||
self.item_service.remove_item(item.id)
|
self.item_service.remove_item(item.id)
|
||||||
self.item_last_use_ms.pop(item.id, None)
|
self.item_last_use_ms.pop(item.id, None)
|
||||||
await self._broadcast(ItemRemovePacket(type="item_remove", itemId=item.id))
|
await self._broadcast(ItemRemovePacket(type="item_remove", itemId=item.id))
|
||||||
self.item_service.save_state()
|
self._request_state_save()
|
||||||
await self._send_item_result(client, True, "delete", f"Deleted {item.title}.", item.id)
|
await self._send_item_result(client, True, "delete", f"Deleted {item.title}.", item.id)
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -1011,7 +1044,7 @@ class SignalingServer:
|
|||||||
await self._send_item_result(client, False, "use", str(exc), item.id)
|
await self._send_item_result(client, False, "use", str(exc), item.id)
|
||||||
return
|
return
|
||||||
item.updatedAt = now_ms
|
item.updatedAt = now_ms
|
||||||
self.item_service.save_state()
|
self._request_state_save()
|
||||||
await self._broadcast_item(item)
|
await self._broadcast_item(item)
|
||||||
|
|
||||||
self.item_last_use_ms[item.id] = now_ms
|
self.item_last_use_ms[item.id] = now_ms
|
||||||
@@ -1200,7 +1233,7 @@ class SignalingServer:
|
|||||||
item.updatedAt = self.item_service.now_ms()
|
item.updatedAt = self.item_service.now_ms()
|
||||||
item.version += 1
|
item.version += 1
|
||||||
await self._broadcast_item(item)
|
await self._broadcast_item(item)
|
||||||
self.item_service.save_state()
|
self._request_state_save()
|
||||||
await self._send_item_result(client, True, "update", f"Updated {item.title}.", item.id)
|
await self._send_item_result(client, True, "update", f"Updated {item.title}.", item.id)
|
||||||
return
|
return
|
||||||
|
|
||||||
|
|||||||
76
server/tests/test_item_schema_ui.py
Normal file
76
server/tests/test_item_schema_ui.py
Normal file
@@ -0,0 +1,76 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from app.server import SignalingServer
|
||||||
|
|
||||||
|
|
||||||
|
def test_ui_definitions_are_complete_for_all_item_types() -> None:
|
||||||
|
server = SignalingServer("127.0.0.1", 8765, None, None)
|
||||||
|
definitions = server._build_ui_definitions()
|
||||||
|
|
||||||
|
item_type_order = definitions.get("itemTypeOrder")
|
||||||
|
item_types = definitions.get("itemTypes")
|
||||||
|
assert isinstance(item_type_order, list)
|
||||||
|
assert isinstance(item_types, list)
|
||||||
|
assert item_type_order
|
||||||
|
assert len(item_types) == len(item_type_order)
|
||||||
|
assert [entry.get("type") for entry in item_types] == item_type_order
|
||||||
|
|
||||||
|
required_global_keys = {
|
||||||
|
"useSound",
|
||||||
|
"emitSound",
|
||||||
|
"useCooldownMs",
|
||||||
|
"emitRange",
|
||||||
|
"directional",
|
||||||
|
"emitSoundSpeed",
|
||||||
|
"emitSoundTempo",
|
||||||
|
}
|
||||||
|
|
||||||
|
for entry in item_types:
|
||||||
|
assert isinstance(entry.get("type"), str)
|
||||||
|
assert isinstance(entry.get("label"), str)
|
||||||
|
assert isinstance(entry.get("editableProperties"), list)
|
||||||
|
assert isinstance(entry.get("propertyMetadata"), dict)
|
||||||
|
assert isinstance(entry.get("propertyOptions"), dict)
|
||||||
|
assert isinstance(entry.get("globalProperties"), dict)
|
||||||
|
|
||||||
|
editable_properties = entry["editableProperties"]
|
||||||
|
property_metadata = entry["propertyMetadata"]
|
||||||
|
property_options = entry["propertyOptions"]
|
||||||
|
global_properties = entry["globalProperties"]
|
||||||
|
|
||||||
|
assert required_global_keys.issubset(set(global_properties.keys()))
|
||||||
|
for property_key in editable_properties:
|
||||||
|
if property_key == "title":
|
||||||
|
continue
|
||||||
|
assert property_key in property_metadata
|
||||||
|
metadata = property_metadata[property_key]
|
||||||
|
assert isinstance(metadata, dict)
|
||||||
|
if metadata.get("valueType") == "list":
|
||||||
|
options = property_options.get(property_key)
|
||||||
|
assert isinstance(options, list)
|
||||||
|
assert options
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_state_save_requests_are_debounced(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
|
server = SignalingServer("127.0.0.1", 8765, None, None)
|
||||||
|
save_calls: list[str] = []
|
||||||
|
|
||||||
|
def fake_save_state() -> None:
|
||||||
|
save_calls.append("saved")
|
||||||
|
|
||||||
|
monkeypatch.setattr(server.item_service, "save_state", fake_save_state)
|
||||||
|
|
||||||
|
server._request_state_save()
|
||||||
|
server._request_state_save()
|
||||||
|
server._request_state_save()
|
||||||
|
await asyncio.sleep(0.25)
|
||||||
|
assert len(save_calls) == 1
|
||||||
|
|
||||||
|
server._request_state_save()
|
||||||
|
await asyncio.sleep(0.25)
|
||||||
|
assert len(save_calls) == 2
|
||||||
Reference in New Issue
Block a user