mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-07 01:13:11 +02:00
refactor: companion FrameServer and related (substantive only, no Black)
Reapply refactor from ce8381a (replace monolithic FrameServer with thin pymc_core subclass, re-export constants, SQLite persistence hooks) while preserving pre-refactor whitespace where patch applied cleanly. Remaining files match refactor commit exactly. Diff vs ce8381a is whitespace-only. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
agessaman
@@ -1,9 +1,9 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from typing import Optional, Dict
|
||||
from typing import Dict, Optional
|
||||
|
||||
from pymc_core.protocol import PacketBuilder, CryptoUtils
|
||||
from pymc_core.protocol import CryptoUtils, PacketBuilder
|
||||
from pymc_core.protocol.constants import PAYLOAD_TYPE_TXT_MSG
|
||||
|
||||
logger = logging.getLogger("RoomServer")
|
||||
@@ -51,7 +51,7 @@ class GlobalRateLimiter:
|
||||
self.min_gap = min_gap_seconds # Minimum gap between consecutive messages
|
||||
self.lock = asyncio.Lock() # Only one transmission at a time
|
||||
self.last_release_time = 0
|
||||
|
||||
|
||||
async def acquire(self):
|
||||
|
||||
async with self.lock:
|
||||
@@ -64,7 +64,7 @@ class GlobalRateLimiter:
|
||||
await asyncio.sleep(wait_time)
|
||||
# Lock is now held - caller can transmit
|
||||
# Will be released when context exits
|
||||
|
||||
|
||||
def release(self):
|
||||
self.last_release_time = time.time()
|
||||
|
||||
@@ -82,43 +82,48 @@ class RoomServer:
|
||||
max_posts: int = 32,
|
||||
config_path: str = None,
|
||||
config: dict = None,
|
||||
config_manager = None,
|
||||
send_advert_callback = None
|
||||
config_manager=None,
|
||||
send_advert_callback=None,
|
||||
):
|
||||
|
||||
|
||||
self.room_hash = room_hash
|
||||
self.room_name = room_name
|
||||
self.local_identity = local_identity
|
||||
self.db = sqlite_handler
|
||||
self.packet_injector = packet_injector
|
||||
self.acl = acl
|
||||
|
||||
|
||||
# Create send_advert callback for this room server
|
||||
async def send_room_advert():
|
||||
"""Send advertisement for this specific room server."""
|
||||
if not packet_injector or not local_identity:
|
||||
logger.error(f"Room '{room_name}': Cannot send advert - missing injector or identity")
|
||||
logger.error(
|
||||
f"Room '{room_name}': Cannot send advert - missing injector or identity"
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
try:
|
||||
from pymc_core.protocol import PacketBuilder
|
||||
from pymc_core.protocol.constants import ADVERT_FLAG_HAS_NAME, ADVERT_FLAG_IS_ROOM_SERVER
|
||||
|
||||
from pymc_core.protocol.constants import (
|
||||
ADVERT_FLAG_HAS_NAME,
|
||||
ADVERT_FLAG_IS_ROOM_SERVER,
|
||||
)
|
||||
|
||||
# Get room config
|
||||
room_config = config.get('identities', {}).get('room_servers', [])
|
||||
room_config = config.get("identities", {}).get("room_servers", [])
|
||||
room_settings = {}
|
||||
for rs in room_config:
|
||||
if rs.get('name') == room_name:
|
||||
room_settings = rs.get('settings', {})
|
||||
if rs.get("name") == room_name:
|
||||
room_settings = rs.get("settings", {})
|
||||
break
|
||||
|
||||
|
||||
# Use room-specific name and location
|
||||
node_name = room_settings.get('room_name', room_name)
|
||||
latitude = room_settings.get('latitude', 0.0)
|
||||
longitude = room_settings.get('longitude', 0.0)
|
||||
|
||||
node_name = room_settings.get("room_name", room_name)
|
||||
latitude = room_settings.get("latitude", 0.0)
|
||||
longitude = room_settings.get("longitude", 0.0)
|
||||
|
||||
flags = ADVERT_FLAG_IS_ROOM_SERVER | ADVERT_FLAG_HAS_NAME
|
||||
|
||||
|
||||
packet = PacketBuilder.create_advert(
|
||||
local_identity=local_identity,
|
||||
name=node_name,
|
||||
@@ -129,21 +134,24 @@ class RoomServer:
|
||||
flags=flags,
|
||||
route_type="flood",
|
||||
)
|
||||
|
||||
|
||||
# Send via packet injector
|
||||
await packet_injector(packet, wait_for_ack=False)
|
||||
|
||||
logger.info(f"Room '{room_name}': Sent flood advert '{node_name}' at ({latitude:.6f}, {longitude:.6f})")
|
||||
|
||||
logger.info(
|
||||
f"Room '{room_name}': Sent flood advert '{node_name}' at ({latitude:.6f}, {longitude:.6f})"
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Room '{room_name}': Failed to send advert: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
# Initialize CLI handler for room server commands
|
||||
self.cli = None
|
||||
if config_path and config and config_manager:
|
||||
from .mesh_cli import MeshCLI
|
||||
|
||||
self.cli = MeshCLI(
|
||||
config_path,
|
||||
config,
|
||||
@@ -152,10 +160,10 @@ class RoomServer:
|
||||
enable_regions=False, # Room servers don't support region commands
|
||||
send_advert_callback=send_room_advert,
|
||||
identity=local_identity,
|
||||
storage_handler=sqlite_handler
|
||||
storage_handler=sqlite_handler,
|
||||
)
|
||||
logger.info(f"Room '{room_name}': Initialized CLI handler with identity and storage")
|
||||
|
||||
|
||||
# Enforce hard limit (match C++ MAX_UNSYNCED_POSTS)
|
||||
if max_posts > MAX_UNSYNCED_POSTS:
|
||||
logger.warning(
|
||||
@@ -164,45 +172,45 @@ class RoomServer:
|
||||
)
|
||||
max_posts = MAX_UNSYNCED_POSTS
|
||||
self.max_posts = max_posts
|
||||
|
||||
|
||||
# Round-robin state
|
||||
self.next_client_idx = 0
|
||||
self.next_push_time = 0
|
||||
|
||||
|
||||
# Cleanup tracking
|
||||
self.last_cleanup_time = time.time()
|
||||
self.cleanup_interval = 600 # Cleanup every 10 minutes
|
||||
|
||||
|
||||
# Safety and monitoring
|
||||
self.client_post_times = {} # Track last N post times per client for rate limiting
|
||||
self.consecutive_sync_errors = 0 # Circuit breaker counter
|
||||
self.last_eviction_check = time.time()
|
||||
self.eviction_check_interval = 300 # Check every 5 minutes
|
||||
|
||||
|
||||
# Initialize global rate limiter (singleton)
|
||||
global _global_push_limiter
|
||||
if _global_push_limiter is None:
|
||||
_global_push_limiter = GlobalRateLimiter(GLOBAL_MIN_GAP_BETWEEN_MESSAGES)
|
||||
self.global_limiter = _global_push_limiter
|
||||
|
||||
|
||||
# Background task handle
|
||||
self._sync_task = None
|
||||
self._running = False
|
||||
|
||||
|
||||
logger.info(
|
||||
f"RoomServer initialized: name='{room_name}', "
|
||||
f"hash=0x{room_hash:02X}, max_posts={max_posts}"
|
||||
)
|
||||
|
||||
|
||||
async def start(self):
|
||||
if self._running:
|
||||
logger.warning(f"Room '{self.room_name}' sync loop already running")
|
||||
return
|
||||
|
||||
|
||||
self._running = True
|
||||
self._sync_task = asyncio.create_task(self._sync_loop())
|
||||
logger.info(f"Room '{self.room_name}' sync loop started")
|
||||
|
||||
|
||||
async def stop(self):
|
||||
self._running = False
|
||||
if self._sync_task:
|
||||
@@ -212,14 +220,14 @@ class RoomServer:
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
logger.info(f"Room '{self.room_name}' sync loop stopped")
|
||||
|
||||
|
||||
async def add_post(
|
||||
self,
|
||||
client_pubkey: bytes,
|
||||
message_text: str,
|
||||
sender_timestamp: int,
|
||||
txt_type: int = TXT_TYPE_PLAIN,
|
||||
allow_server_author: bool = False
|
||||
allow_server_author: bool = False,
|
||||
) -> bool:
|
||||
|
||||
try:
|
||||
@@ -230,20 +238,19 @@ class RoomServer:
|
||||
f"exceeds max length ({len(message_text)} > {MAX_MESSAGE_LENGTH}), truncating"
|
||||
)
|
||||
message_text = message_text[:MAX_MESSAGE_LENGTH]
|
||||
|
||||
|
||||
# SAFETY: Rate limit per client
|
||||
client_key = client_pubkey.hex()
|
||||
now = time.time()
|
||||
|
||||
|
||||
if client_key not in self.client_post_times:
|
||||
self.client_post_times[client_key] = []
|
||||
|
||||
|
||||
# Remove timestamps older than 1 minute
|
||||
self.client_post_times[client_key] = [
|
||||
t for t in self.client_post_times[client_key]
|
||||
if now - t < 60
|
||||
t for t in self.client_post_times[client_key] if now - t < 60
|
||||
]
|
||||
|
||||
|
||||
# Check rate limit
|
||||
if len(self.client_post_times[client_key]) >= MAX_POSTS_PER_CLIENT_PER_MINUTE:
|
||||
logger.warning(
|
||||
@@ -251,13 +258,13 @@ class RoomServer:
|
||||
f"exceeded rate limit ({MAX_POSTS_PER_CLIENT_PER_MINUTE} posts/min), dropping message"
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
# Record this post time
|
||||
self.client_post_times[client_key].append(now)
|
||||
|
||||
|
||||
# Use our RTC time for post_timestamp
|
||||
post_timestamp = time.time()
|
||||
|
||||
|
||||
# Store to database
|
||||
msg_id = self.db.insert_room_message(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
@@ -265,22 +272,22 @@ class RoomServer:
|
||||
message_text=message_text,
|
||||
post_timestamp=post_timestamp,
|
||||
sender_timestamp=sender_timestamp,
|
||||
txt_type=txt_type
|
||||
txt_type=txt_type,
|
||||
)
|
||||
|
||||
|
||||
if msg_id:
|
||||
logger.info(
|
||||
f"Room '{self.room_name}': New post #{msg_id} from "
|
||||
f"{client_pubkey[:4].hex()}: {message_text[:50]}"
|
||||
)
|
||||
|
||||
|
||||
# Log authenticated clients count for debugging distribution
|
||||
all_clients = self.acl.get_all_clients()
|
||||
logger.info(
|
||||
f"Room '{self.room_name}': Message stored, will distribute to "
|
||||
f"{len(all_clients)} authenticated client(s)"
|
||||
)
|
||||
|
||||
|
||||
# Update client's sync_since to this message's timestamp
|
||||
# This prevents the author from receiving their own message back
|
||||
# Also update activity timestamp (they're clearly active if posting)
|
||||
@@ -292,43 +299,43 @@ class RoomServer:
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client_pubkey.hex(),
|
||||
sync_since=post_timestamp, # Don't send this message back to author
|
||||
last_activity=time.time()
|
||||
last_activity=time.time(),
|
||||
)
|
||||
|
||||
|
||||
# Trigger push notification
|
||||
self.next_push_time = time.time() + (PUSH_NOTIFY_DELAY_MS / 1000.0)
|
||||
|
||||
|
||||
return True
|
||||
else:
|
||||
logger.error(f"Failed to store message to database")
|
||||
return False
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding post: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
async def push_post_to_client(self, client_info, post: Dict) -> bool:
|
||||
|
||||
|
||||
try:
|
||||
# SAFETY: Global transmission lock - only ONE message on radio at a time
|
||||
# This is critical because LoRa is serial (0.5-9s airtime per message)
|
||||
await self.global_limiter.acquire()
|
||||
|
||||
|
||||
# SAFETY: Check client failure backoff
|
||||
sync_state = self.db.get_client_sync(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client_info.id.get_public_key().hex()
|
||||
client_pubkey=client_info.id.get_public_key().hex(),
|
||||
)
|
||||
|
||||
|
||||
if sync_state:
|
||||
failures = sync_state.get('push_failures', 0)
|
||||
failures = sync_state.get("push_failures", 0)
|
||||
if failures > 0:
|
||||
# Apply exponential backoff
|
||||
backoff_idx = min(failures, len(RETRY_BACKOFF_SCHEDULE) - 1)
|
||||
backoff_delay = RETRY_BACKOFF_SCHEDULE[backoff_idx]
|
||||
last_failure_time = sync_state.get('updated_at', 0)
|
||||
last_failure_time = sync_state.get("updated_at", 0)
|
||||
time_since_failure = time.time() - last_failure_time
|
||||
|
||||
|
||||
if time_since_failure < backoff_delay:
|
||||
wait_time = backoff_delay - time_since_failure
|
||||
logger.debug(
|
||||
@@ -336,33 +343,30 @@ class RoomServer:
|
||||
f"in backoff (failure {failures}), waiting {wait_time:.0f}s"
|
||||
)
|
||||
return False # Skip this client for now
|
||||
|
||||
|
||||
# Build message payload
|
||||
timestamp = int(time.time())
|
||||
flags = (TXT_TYPE_SIGNED_PLAIN << 2) # Include author prefix
|
||||
|
||||
flags = TXT_TYPE_SIGNED_PLAIN << 2 # Include author prefix
|
||||
|
||||
# Author prefix (first 4 bytes of pubkey)
|
||||
author_pubkey = bytes.fromhex(post['author_pubkey'])
|
||||
author_pubkey = bytes.fromhex(post["author_pubkey"])
|
||||
author_prefix = author_pubkey[:4]
|
||||
|
||||
|
||||
# Plaintext: timestamp(4) + flags(1) + author_prefix(4) + text
|
||||
message_bytes = post['message_text'].encode('utf-8')
|
||||
message_bytes = post["message_text"].encode("utf-8")
|
||||
plaintext = (
|
||||
timestamp.to_bytes(4, 'little') +
|
||||
bytes([flags]) +
|
||||
author_prefix +
|
||||
message_bytes
|
||||
timestamp.to_bytes(4, "little") + bytes([flags]) + author_prefix + message_bytes
|
||||
)
|
||||
|
||||
|
||||
# Calculate expected ACK (same algorithm as pymc_core)
|
||||
attempt = 0
|
||||
pack_data = PacketBuilder._pack_timestamp_data(timestamp, attempt, message_bytes)
|
||||
ack_hash = CryptoUtils.sha256(pack_data + client_info.id.get_public_key())[:4]
|
||||
expected_ack_crc = int.from_bytes(ack_hash, 'little')
|
||||
|
||||
expected_ack_crc = int.from_bytes(ack_hash, "little")
|
||||
|
||||
# Determine routing based on stored out_path
|
||||
route_type = "flood" if client_info.out_path_len < 0 else "direct"
|
||||
|
||||
|
||||
# Create datagram
|
||||
packet = PacketBuilder.create_datagram(
|
||||
ptype=PAYLOAD_TYPE_TXT_MSG,
|
||||
@@ -370,41 +374,42 @@ class RoomServer:
|
||||
local_identity=self.local_identity,
|
||||
secret=client_info.shared_secret,
|
||||
plaintext=plaintext,
|
||||
route_type=route_type
|
||||
route_type=route_type,
|
||||
)
|
||||
|
||||
|
||||
# Add stored path for direct routing
|
||||
if route_type == "direct" and len(client_info.out_path) > 0:
|
||||
packet.path = bytearray(client_info.out_path[:client_info.out_path_len])
|
||||
packet.path = bytearray(client_info.out_path[: client_info.out_path_len])
|
||||
packet.path_len = client_info.out_path_len
|
||||
|
||||
|
||||
# Calculate ACK timeout
|
||||
if route_type == "flood":
|
||||
ack_timeout = PUSH_ACK_TIMEOUT_FLOOD_MS / 1000.0
|
||||
else:
|
||||
path_len = client_info.out_path_len if client_info.out_path_len >= 0 else 0
|
||||
ack_timeout = (PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (path_len + 1)) / 1000.0
|
||||
|
||||
ack_timeout = (
|
||||
PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (path_len + 1)
|
||||
) / 1000.0
|
||||
|
||||
# Update client sync state with pending ACK
|
||||
self.db.upsert_client_sync(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client_info.id.get_public_key().hex(),
|
||||
pending_ack_crc=expected_ack_crc,
|
||||
push_post_timestamp=post['post_timestamp'],
|
||||
ack_timeout_time=time.time() + ack_timeout
|
||||
push_post_timestamp=post["post_timestamp"],
|
||||
ack_timeout_time=time.time() + ack_timeout,
|
||||
)
|
||||
# Send packet (dispatcher will track ACK automatically)
|
||||
# This blocks for the entire transmission duration (0.5-9 seconds)
|
||||
success = await self.packet_injector(packet, wait_for_ack=True)
|
||||
|
||||
|
||||
# SAFETY: Release transmission lock AFTER send completes
|
||||
self.global_limiter.release()
|
||||
|
||||
|
||||
if success:
|
||||
# ACK received! Update sync state
|
||||
await self._handle_ack_received(
|
||||
client_info.id.get_public_key(),
|
||||
post['post_timestamp']
|
||||
client_info.id.get_public_key(), post["post_timestamp"]
|
||||
)
|
||||
logger.info(
|
||||
f"Room '{self.room_name}': Pushed post to "
|
||||
@@ -417,13 +422,13 @@ class RoomServer:
|
||||
f"Room '{self.room_name}': Push to "
|
||||
f"0x{client_info.id.get_public_key()[0]:02X} timed out"
|
||||
)
|
||||
|
||||
|
||||
return success
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error pushing post to client: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
async def _handle_ack_received(self, client_pubkey: bytes, post_timestamp: float):
|
||||
|
||||
try:
|
||||
@@ -434,29 +439,28 @@ class RoomServer:
|
||||
sync_since=post_timestamp,
|
||||
pending_ack_crc=0,
|
||||
push_failures=0,
|
||||
last_activity=time.time()
|
||||
last_activity=time.time(),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error handling ACK received: {e}")
|
||||
|
||||
|
||||
async def _handle_ack_timeout(self, client_pubkey: bytes):
|
||||
try:
|
||||
# Get current sync state
|
||||
sync_state = self.db.get_client_sync(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client_pubkey.hex()
|
||||
room_hash=f"0x{self.room_hash:02X}", client_pubkey=client_pubkey.hex()
|
||||
)
|
||||
|
||||
|
||||
if sync_state:
|
||||
# Increment failure counter, clear pending_ack
|
||||
failures = sync_state.get('push_failures', 0) + 1
|
||||
failures = sync_state.get("push_failures", 0) + 1
|
||||
self.db.upsert_client_sync(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client_pubkey.hex(),
|
||||
push_failures=failures,
|
||||
pending_ack_crc=0
|
||||
pending_ack_crc=0,
|
||||
)
|
||||
|
||||
|
||||
if failures >= 3:
|
||||
logger.warning(
|
||||
f"Room '{self.room_name}': Client 0x{client_pubkey[0]:02X} "
|
||||
@@ -464,86 +468,86 @@ class RoomServer:
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error handling ACK timeout: {e}")
|
||||
|
||||
|
||||
def get_unsynced_count(self, client_pubkey: bytes) -> int:
|
||||
try:
|
||||
# Get client's sync state
|
||||
sync_state = self.db.get_client_sync(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client_pubkey.hex()
|
||||
room_hash=f"0x{self.room_hash:02X}", client_pubkey=client_pubkey.hex()
|
||||
)
|
||||
|
||||
sync_since = sync_state['sync_since'] if sync_state else 0
|
||||
|
||||
|
||||
sync_since = sync_state["sync_since"] if sync_state else 0
|
||||
|
||||
return self.db.get_unsynced_count(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client_pubkey.hex(),
|
||||
sync_since=sync_since
|
||||
sync_since=sync_since,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting unsynced count: {e}")
|
||||
return 0
|
||||
|
||||
|
||||
async def _evict_failed_clients(self):
|
||||
try:
|
||||
now = time.time()
|
||||
all_sync_states = self.db.get_all_room_clients(f"0x{self.room_hash:02X}")
|
||||
|
||||
|
||||
for sync_state in all_sync_states:
|
||||
client_pubkey_hex = sync_state['client_pubkey']
|
||||
push_failures = sync_state.get('push_failures', 0)
|
||||
last_activity = sync_state.get('last_activity', 0)
|
||||
|
||||
client_pubkey_hex = sync_state["client_pubkey"]
|
||||
push_failures = sync_state.get("push_failures", 0)
|
||||
last_activity = sync_state.get("last_activity", 0)
|
||||
|
||||
# Skip already-evicted clients (marked with last_activity=0)
|
||||
if last_activity == 0:
|
||||
continue
|
||||
|
||||
|
||||
evict = False
|
||||
reason = ""
|
||||
|
||||
|
||||
# Check max failures
|
||||
if push_failures >= MAX_PUSH_FAILURES:
|
||||
evict = True
|
||||
reason = f"max failures ({push_failures})"
|
||||
|
||||
|
||||
# Check inactivity timeout
|
||||
elif now - last_activity > INACTIVE_CLIENT_TIMEOUT:
|
||||
evict = True
|
||||
reason = f"inactive for {(now - last_activity) / 60:.0f} minutes"
|
||||
|
||||
|
||||
if evict:
|
||||
# Remove from database
|
||||
self.db.upsert_client_sync(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client_pubkey_hex,
|
||||
last_activity=0 # Mark as evicted
|
||||
last_activity=0, # Mark as evicted
|
||||
)
|
||||
|
||||
|
||||
# Remove from ACL
|
||||
client_pubkey = bytes.fromhex(client_pubkey_hex)
|
||||
self.acl.remove_client(client_pubkey)
|
||||
|
||||
|
||||
logger.info(
|
||||
f"Room '{self.room_name}': Evicted client "
|
||||
f"0x{client_pubkey[0]:02X} ({reason})"
|
||||
)
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error evicting failed clients: {e}", exc_info=True)
|
||||
|
||||
|
||||
async def _sync_loop(self):
|
||||
|
||||
# SAFETY: Stagger room startup to prevent thundering herd
|
||||
import random
|
||||
|
||||
startup_delay = random.uniform(0, 5) # 0-5 second random delay
|
||||
await asyncio.sleep(startup_delay)
|
||||
|
||||
|
||||
logger.info(f"Room '{self.room_name}' sync loop starting (delayed {startup_delay:.1f}s)")
|
||||
|
||||
|
||||
while self._running:
|
||||
try:
|
||||
await asyncio.sleep(SYNC_PUSH_INTERVAL_MS / 1000.0)
|
||||
|
||||
|
||||
# SAFETY: Circuit breaker - stop if too many consecutive errors
|
||||
if self.consecutive_sync_errors >= MAX_CONSECUTIVE_SYNC_ERRORS:
|
||||
logger.error(
|
||||
@@ -553,21 +557,21 @@ class RoomServer:
|
||||
await asyncio.sleep(DB_ERROR_RETRY_DELAY)
|
||||
self.consecutive_sync_errors = 0 # Reset after pause
|
||||
continue
|
||||
|
||||
|
||||
# SAFETY: Periodic eviction check (every 5 minutes)
|
||||
if time.time() - self.last_eviction_check > self.eviction_check_interval:
|
||||
await self._evict_failed_clients()
|
||||
self.last_eviction_check = time.time()
|
||||
|
||||
|
||||
# Periodic cleanup check (every 10 minutes)
|
||||
if time.time() - self.last_cleanup_time > self.cleanup_interval:
|
||||
await self._cleanup_old_messages()
|
||||
self.last_cleanup_time = time.time()
|
||||
|
||||
|
||||
# Check if it's time to push
|
||||
if time.time() < self.next_push_time:
|
||||
continue
|
||||
|
||||
|
||||
# Get all clients for this room
|
||||
all_clients = self.acl.get_all_clients()
|
||||
if not all_clients:
|
||||
@@ -575,60 +579,66 @@ class RoomServer:
|
||||
# to avoid log spam when room is idle
|
||||
self.next_push_time = time.time() + 1.0 # Check again in 1 second
|
||||
continue
|
||||
|
||||
|
||||
# SAFETY: Limit number of clients
|
||||
if len(all_clients) > MAX_CLIENTS_PER_ROOM:
|
||||
logger.warning(
|
||||
f"Room '{self.room_name}': Too many clients ({len(all_clients)} > {MAX_CLIENTS_PER_ROOM})"
|
||||
)
|
||||
all_clients = all_clients[:MAX_CLIENTS_PER_ROOM]
|
||||
|
||||
|
||||
# Check for ACK timeouts first
|
||||
await self._check_ack_timeouts()
|
||||
|
||||
|
||||
# Track how many clients we've checked in this iteration
|
||||
clients_checked = 0
|
||||
max_checks = len(all_clients)
|
||||
|
||||
|
||||
# Round-robin: find next active client
|
||||
while clients_checked < max_checks:
|
||||
# Get next client
|
||||
if self.next_client_idx >= len(all_clients):
|
||||
self.next_client_idx = 0
|
||||
|
||||
|
||||
client = all_clients[self.next_client_idx]
|
||||
self.next_client_idx = (self.next_client_idx + 1) % len(all_clients)
|
||||
clients_checked += 1
|
||||
|
||||
|
||||
# Get client sync state
|
||||
sync_state = self.db.get_client_sync(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client.id.get_public_key().hex()
|
||||
client_pubkey=client.id.get_public_key().hex(),
|
||||
)
|
||||
|
||||
|
||||
# Skip if already waiting for ACK, evicted, or max failures
|
||||
if sync_state:
|
||||
pending_ack = sync_state.get('pending_ack_crc', 0)
|
||||
last_activity = sync_state.get('last_activity', 0)
|
||||
push_failures = sync_state.get('push_failures', 0)
|
||||
|
||||
pending_ack = sync_state.get("pending_ack_crc", 0)
|
||||
last_activity = sync_state.get("last_activity", 0)
|
||||
push_failures = sync_state.get("push_failures", 0)
|
||||
|
||||
if pending_ack != 0:
|
||||
logger.debug(f"Skipping client 0x{client.id.get_public_key()[0]:02X} (waiting for ACK)")
|
||||
logger.debug(
|
||||
f"Skipping client 0x{client.id.get_public_key()[0]:02X} (waiting for ACK)"
|
||||
)
|
||||
continue
|
||||
|
||||
|
||||
if last_activity == 0:
|
||||
logger.debug(f"Skipping client 0x{client.id.get_public_key()[0]:02X} (evicted)")
|
||||
logger.debug(
|
||||
f"Skipping client 0x{client.id.get_public_key()[0]:02X} (evicted)"
|
||||
)
|
||||
continue
|
||||
|
||||
|
||||
if push_failures >= 3:
|
||||
logger.debug(f"Skipping client 0x{client.id.get_public_key()[0]:02X} (max failures)")
|
||||
logger.debug(
|
||||
f"Skipping client 0x{client.id.get_public_key()[0]:02X} (max failures)"
|
||||
)
|
||||
continue
|
||||
|
||||
sync_since = sync_state.get('sync_since', 0)
|
||||
|
||||
sync_since = sync_state.get("sync_since", 0)
|
||||
else:
|
||||
# Initialize sync state for new client
|
||||
# Use sync_since from ACL client (sent during login) if available
|
||||
sync_since = client.sync_since if hasattr(client, 'sync_since') else 0
|
||||
sync_since = client.sync_since if hasattr(client, "sync_since") else 0
|
||||
logger.info(
|
||||
f"Room '{self.room_name}': Initializing client "
|
||||
f"0x{client.id.get_public_key()[0]:02X} with sync_since={sync_since}"
|
||||
@@ -637,17 +647,17 @@ class RoomServer:
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client.id.get_public_key().hex(),
|
||||
sync_since=sync_since,
|
||||
last_activity=time.time()
|
||||
last_activity=time.time(),
|
||||
)
|
||||
|
||||
|
||||
# Find next unsynced message for this client
|
||||
unsynced = self.db.get_unsynced_messages(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
client_pubkey=client.id.get_public_key().hex(),
|
||||
sync_since=sync_since,
|
||||
limit=1
|
||||
limit=1,
|
||||
)
|
||||
|
||||
|
||||
if unsynced:
|
||||
post = unsynced[0]
|
||||
logger.debug(
|
||||
@@ -656,7 +666,7 @@ class RoomServer:
|
||||
)
|
||||
# Check if enough time has passed since post creation
|
||||
now = time.time()
|
||||
if now >= post['post_timestamp'] + POST_SYNC_DELAY_SECS:
|
||||
if now >= post["post_timestamp"] + POST_SYNC_DELAY_SECS:
|
||||
# Push this post
|
||||
await self.push_post_to_client(client, post)
|
||||
self.next_push_time = time.time() + (SYNC_PUSH_INTERVAL_MS / 1000.0)
|
||||
@@ -668,15 +678,15 @@ class RoomServer:
|
||||
else:
|
||||
# No unsynced posts for this client, try next client
|
||||
continue
|
||||
|
||||
|
||||
# If we checked all clients and none were active/ready
|
||||
if clients_checked >= max_checks:
|
||||
# All clients skipped or no messages - wait longer before next check
|
||||
self.next_push_time = time.time() + 5.0 # Wait 5 seconds
|
||||
|
||||
|
||||
# SAFETY: Reset error counter on successful iteration
|
||||
self.consecutive_sync_errors = 0
|
||||
|
||||
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
except Exception as e:
|
||||
@@ -684,35 +694,34 @@ class RoomServer:
|
||||
self.consecutive_sync_errors += 1
|
||||
logger.error(
|
||||
f"Room '{self.room_name}': Sync loop error #{self.consecutive_sync_errors}: {e}",
|
||||
exc_info=True
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
|
||||
# SAFETY: Back off on errors
|
||||
backoff = min(self.consecutive_sync_errors, 10) # Cap at 10 seconds
|
||||
await asyncio.sleep(backoff)
|
||||
|
||||
|
||||
logger.info(f"Room '{self.room_name}' sync loop stopped")
|
||||
|
||||
|
||||
async def _check_ack_timeouts(self):
|
||||
try:
|
||||
now = time.time()
|
||||
all_sync_states = self.db.get_all_room_clients(f"0x{self.room_hash:02X}")
|
||||
|
||||
|
||||
for sync_state in all_sync_states:
|
||||
if sync_state['pending_ack_crc'] != 0:
|
||||
timeout_time = sync_state.get('ack_timeout_time', 0)
|
||||
if sync_state["pending_ack_crc"] != 0:
|
||||
timeout_time = sync_state.get("ack_timeout_time", 0)
|
||||
if now >= timeout_time:
|
||||
# ACK timeout
|
||||
client_pubkey = bytes.fromhex(sync_state['client_pubkey'])
|
||||
client_pubkey = bytes.fromhex(sync_state["client_pubkey"])
|
||||
await self._handle_ack_timeout(client_pubkey)
|
||||
except Exception as e:
|
||||
logger.error(f"Error checking ACK timeouts: {e}")
|
||||
|
||||
|
||||
async def _cleanup_old_messages(self):
|
||||
try:
|
||||
deleted = self.db.cleanup_old_messages(
|
||||
room_hash=f"0x{self.room_hash:02X}",
|
||||
keep_count=self.max_posts
|
||||
room_hash=f"0x{self.room_hash:02X}", keep_count=self.max_posts
|
||||
)
|
||||
if deleted > 0:
|
||||
logger.info(f"Room '{self.room_name}': Cleaned up {deleted} old messages")
|
||||
|
||||
Reference in New Issue
Block a user