""" K1RL_QU1NT V10.2 - Redis Connection Manager (HuggingFace Spaces Edition) ========================================================================= Production-grade replacement for DedicatedAblyConnectionManager. Optimized for container environments and HuggingFace Spaces. LATENCY FIXES (V10.0): V9.0 PROBLEM: - pubsub.get_message(timeout=0.1) → 100ms polling delay - command_queue.get(timeout=1.0) → 1000ms polling delay - Total worst-case latency: ~1100ms V10.1 SOLUTION (HF Spaces): - pubsub.get_message(timeout=None) → TRUE BLOCKING (0ms latency) - threading.Event() for commands → INSTANT signaling (0ms latency) - Separate threads for pub vs sub → No blocking interference - Container-safe Redis configuration - Environment-aware connection settings - Expected latency: <1ms (network RTT only) LATEST-STATE LATCH MODEL (V10.2 — transport-side coalescing): PROBLEM: Redis pub/sub is an ordered transport, NOT a latest-state store. When QUASAR is busy processing one bound state, subsequent feature/state messages accumulate and the consumer is forced to walk through every stale intermediate state one-by-one before reaching the current one. This causes the live trading path to replay old feature transactions that are already superseded by the time they are processed. V10.2 SOLUTION (latest-state latch): For hot-path live-state channels (feature vectors, bound state updates), subscribe with latest_only=True. The manager then maintains a single pending-latest slot per channel inside the manager process: ARRIVING MSG ──► consumer BUSY? ──YES──► overwrite pending_latest[ch] │ (previous pending discarded) NO │ ▼ fire callback immediately mark in_flight[ch] = True │ callback returns │ pending_latest[ch] exists? YES ──► dispatch freshest, clear slot, repeat NO ──► in_flight[ch] = False (go idle) Net result: QUASAR sees at most ONE waiting message per channel after each processing cycle — always the freshest available state. SCOPE: - latest_only=True → latch model (hot-path live state, feature channels) - latest_only=False → lossless queue (rewards, audit/replay channels — UNCHANGED) DIAGNOSTICS: manager.latch_stats → coalescing counters (property) manager.get_latch_diagnostics() → detailed in-flight + pending inspection manager.get_state() → includes 'latch_model' sub-dict ARCHITECTURE: ┌─────────────────────────────────────────────────────────────────┐ │ Main Application Thread │ │ subscribe() ──┬──► CommandQueue ──► CommandThread (non-block) │ │ publish() ──┘ │ └─────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ ListenerThread (BLOCKING) │ │ pubsub.listen() ──► Yield messages instantly │ │ │ │ │ ├─ latest_only channel ──► latch model (coalesce-to-freshest)│ │ └─ normal channel ──► direct callback (lossless) │ └─────────────────────────────────────────────────────────────────┘ HUGGINGFACE SPACES OPTIMIZATIONS: - Auto-detect container environment - Use redis_config for connection settings - Container-safe logging and paths - Resource-optimized connection pools Author: K1RL_QU1NT System Version: 10.2.0 (Latest-State Latch Edition) """ import asyncio import json import queue import threading import time import logging import weakref import os from datetime import datetime from typing import Optional, Callable, Dict, Any, List, Union from dataclasses import dataclass, field from contextlib import contextmanager from enum import Enum, auto import redis import redis.asyncio as aioredis # ✅ V75: Import redis_config_v75 for HuggingFace Spaces compatibility try: from redis_config_v75 import REDIS_URL, REDIS_PASSWORD, REDIS_DB_FEATURES, REDIS_DB_REWARDS HAS_REDIS_CONFIG = True except ImportError: # Fallback for development/testing REDIS_URL = "redis://localhost:6379/0" REDIS_PASSWORD = os.environ.get("REDIS_PASSWORD", "k1rl_v75_8d2f5a1c9e4b7063") # FIX: Use actual password instead of None REDIS_DB_FEATURES = 0 REDIS_DB_REWARDS = 1 HAS_REDIS_CONFIG = False logger = logging.getLogger(__name__) # ✅ ADDED: HuggingFace Spaces environment detection IS_HF_SPACES = os.environ.get('SPACE_ID') is not None # ============================================================================= # CONSTANTS & CONFIGURATION (HuggingFace Spaces Edition) # ============================================================================= class ConnectionState(Enum): """Connection state enum (mirrors Ably states).""" INITIALIZED = auto() CONNECTING = auto() CONNECTED = auto() DISCONNECTED = auto() SUSPENDED = auto() CLOSING = auto() CLOSED = auto() FAILED = auto() @dataclass class RedisConfig: """Centralized configuration for Redis connection (HuggingFace Spaces optimized).""" # ✅ FIXED: Use redis_config URL with fallback url: str = REDIS_URL password: Optional[str] = REDIS_PASSWORD socket_timeout: float = 10.0 if not IS_HF_SPACES else 15.0 # Longer timeout for HF Spaces socket_connect_timeout: float = 10.0 if not IS_HF_SPACES else 15.0 health_check_interval: int = 15 if not IS_HF_SPACES else 30 # Less frequent for containers retry_on_timeout: bool = True decode_responses: bool = True # Reconnection settings (container-optimized) reconnect_base_interval: float = 0.5 if not IS_HF_SPACES else 2.0 # Slower for containers reconnect_max_interval: float = 30.0 if not IS_HF_SPACES else 60.0 # Longer max for containers reconnect_max_attempts: Optional[int] = None # None = infinite # Stream settings (optional persistence) use_streams: bool = False stream_maxlen: int = 10000 @classmethod def from_url(cls, url: str, **kwargs) -> 'RedisConfig': return cls(url=url, **kwargs) @classmethod def for_database(cls, db_number: int, **kwargs) -> 'RedisConfig': """Create config for specific database number""" base_url = REDIS_URL.rsplit('/', 1)[0] if '/' in REDIS_URL else REDIS_URL url = f"{base_url}/{db_number}" return cls(url=url, **kwargs) # ============================================================================= # MESSAGE WRAPPER (Ably message compatibility) # ============================================================================= @dataclass class RedisMessage: """ Mimics ably.types.message.Message interface. Ably messages have: .name (event), .data (payload), .timestamp, .id This wrapper provides the same attributes so existing callbacks work unchanged. """ channel: str name: str # event name (Ably: message.name) data: Any # payload (Ably: message.data) timestamp: float = field(default_factory=time.time) # Computed properties @property def id(self) -> str: return f"{self.channel}:{int(self.timestamp * 1000)}" # Ably compatibility attributes encoding: Optional[str] = None client_id: Optional[str] = None connection_id: Optional[str] = None def __repr__(self): data_preview = list(self.data.keys())[:3] if isinstance(self.data, dict) else '...' return f"RedisMessage(channel={self.channel}, event={self.name}, data_keys={data_preview})" # ============================================================================= # COMMAND TYPES (Type-safe command passing) # ============================================================================= @dataclass class Command: """Base command class for thread-safe operations.""" pass @dataclass class SubscribeCommand(Command): channel: str callback: Optional[Callable] = None event_filter: Optional[str] = None # Filter specific events latest_only: bool = False # Enable latest-state latch mode for this channel @dataclass class UnsubscribeCommand(Command): channel: str @dataclass class PublishCommand(Command): channel: str message: dict event_name: str = "message" response_event: Optional[threading.Event] = None result: Optional[int] = None # Number of receivers @dataclass class GetStateCommand(Command): response_event: threading.Event = field(default_factory=threading.Event) result: Optional[dict] = None @dataclass class StopCommand(Command): pass # ============================================================================= # ZERO-LATENCY REDIS CONNECTION MANAGER (HuggingFace Spaces Edition) # ============================================================================= class DedicatedRedisConnectionManager: """ V10.1: Zero-latency Redis-based replacement for DedicatedAblyConnectionManager. HuggingFace Spaces optimized with container-safe configuration. KEY IMPROVEMENTS OVER V10.0: 1. HuggingFace Spaces environment detection 2. redis_config integration for consistent connection settings 3. Container-optimized timeouts and intervals 4. Enhanced error handling for container networking 5. Resource-efficient connection pooling IDENTICAL PUBLIC API: .start() → Start background threads .subscribe(channel, callback) → Subscribe to channel .publish(channel, message, event) → Publish message .get_state(timeout) → Get connection state .stop() → Stop manager .connected → Bool connection state .stats → Dict of statistics """ def __init__( self, redis_url: Optional[str] = None, # ✅ FIXED: Optional, uses redis_config password: Optional[str] = None, on_connected: Optional[Callable] = None, on_disconnected: Optional[Callable] = None, on_message: Optional[Callable] = None, reconnect_interval: Optional[float] = None, max_reconnect_attempts: Optional[int] = None, use_streams: bool = False, stream_maxlen: int = 10000, database: int = 0, # ✅ ADDED: Database selection # Legacy Ably params (accepted but ignored for drop-in compat) api_key: str = None, ): # ✅ FIXED: Use redis_config with environment awareness if redis_url is None: if database == REDIS_DB_FEATURES: config = RedisConfig.for_database(REDIS_DB_FEATURES) elif database == REDIS_DB_REWARDS: config = RedisConfig.for_database(REDIS_DB_REWARDS) else: config = RedisConfig.for_database(database) else: config = RedisConfig.from_url(redis_url) # Override with provided parameters if password is not None: config.password = password if reconnect_interval is not None: config.reconnect_base_interval = reconnect_interval if max_reconnect_attempts is not None: config.reconnect_max_attempts = max_reconnect_attempts config.use_streams = use_streams config.stream_maxlen = stream_maxlen self.config = config # Callbacks self.on_connected = on_connected self.on_disconnected = on_disconnected self.on_message = on_message if api_key: logger.info(f"ℹ️ [DedicatedRedisConnectionManager] Ably API key ignored (migration compatibility)") # Connection management self._redis: Optional[redis.Redis] = None self._pubsub: Optional[redis.client.PubSub] = None self._state = ConnectionState.INITIALIZED self._connected = False self._running = False # Thread management self._listener_thread: Optional[threading.Thread] = None self._command_thread: Optional[threading.Thread] = None self._command_queue: queue.Queue = queue.Queue() self._shutdown_event = threading.Event() # Subscription management self._subscriptions: Dict[str, Callable] = {} self._subscription_lock = threading.RLock() # ========================================================================= # LATEST-STATE LATCH MODEL (V10.2 — transport-side coalescing) # # DESIGN INTENT: # Redis pub/sub is an ordered transport, NOT a latest-state store. # Every published message arrives at this listener in arrival order. # For hot-path live-state channels (e.g. features/bound-state), QUASAR # does not need every intermediate state — it needs only the CURRENT state # when the previous processing cycle finishes. # # This latch model implements collapse-to-freshest INSIDE the manager: # - When the consumer is marked busy (in_flight[ch] == True), any new # arrival for that channel overwrites pending_latest[ch] rather than # enqueueing another callback dispatch. # - When the consumer clears busy, exactly ONE dispatch fires (the newest # pending state), and all superseded messages are silently dropped. # - Reward/durable channels that are NOT registered as latest_only are # completely unaffected — they remain lossless queue-style consumers. # # SCOPE: # Applies ONLY to channels subscribed with latest_only=True. # Reward channels, audit/replay channels, and stream consumers use the # original dispatch path with no coalescing. # ========================================================================= # Set of channels operating in latest_only mode self._latest_only_channels: set = set() self._latch_lock = threading.RLock() # Per-channel busy flag: True while the callback for that channel is executing self._in_flight: Dict[str, bool] = {} # Per-channel pending-latest slot: holds the single newest unprocessed message # Any subsequent arrival overwrites this slot (overwrite is intentional) self._pending_latest: Dict[str, Any] = {} # Coalescing diagnostics self._latch_stats: Dict[str, int] = { 'messages_overwritten': 0, # How many arrivals replaced a waiting pending slot 'messages_dispatched': 0, # How many latest-only callbacks were actually fired 'busy_drops_avoided': 0, # Callbacks NOT fired because consumer was busy (coalesced) 'pending_slot_occupancy': 0, # Instantaneous count of occupied pending slots (updated on read) } self._latch_stats_lock = threading.RLock() # ✅ STREAM CATCHUP: Ably-inspired message recovery via Redis Streams # Pub/sub for real-time (low latency), Streams for durability (zero message loss) # Inspired by Ably's timeserial tracking + 2-min recovery window self._stream_consumer_group = f"quasar-{int(time.time())}" self._stream_consumer_name = f"consumer-{os.getpid()}" self._stream_last_ids: Dict[str, str] = {} # Track last processed stream ID per channel self._stream_dedup: set = set() # Dedup window (msg timestamps seen via pub/sub) self._stream_dedup_max = 5000 self._stream_channels: List[str] = [] # Channels to read from streams # Statistics self._stats = { 'connections': 0, 'disconnections': 0, 'reconnections': 0, 'messages_received': 0, 'messages_published': 0, 'publish_errors': 0, 'last_connected': None, 'total_uptime': 0, 'avg_latency_ms': 0, } self._stats_lock = threading.RLock() self._start_time = time.time() if IS_HF_SPACES: logger.info("🤗 HuggingFace Spaces environment detected - using container-optimized settings") logger.info(f"✅ DedicatedRedisConnectionManager V10.1 initialized") logger.info(f" Redis URL: {self.config.url}") logger.info(f" Database: {database}") logger.info(f" HF Spaces: {IS_HF_SPACES}") def start(self, timeout: float = 10.0) -> bool: """Start the connection manager with HuggingFace Spaces compatibility.""" if self._running: logger.warning("Manager already running") return True try: logger.info("🚀 Starting DedicatedRedisConnectionManager V10.1...") # Longer timeout for HF Spaces actual_timeout = timeout * 2 if IS_HF_SPACES else timeout # Initialize Redis connections if not self._connect(timeout=actual_timeout): logger.error("❌ Failed to establish Redis connection") return False # Start background threads self._running = True self._start_threads() # Wait for connection confirmation start_time = time.time() while not self._connected and (time.time() - start_time) < actual_timeout: time.sleep(0.1) if self._connected: logger.info(f"✅ DedicatedRedisConnectionManager started successfully") if self.on_connected: self.on_connected() return True else: logger.error("❌ Failed to confirm connection within timeout") self.stop() return False except Exception as e: logger.error(f"❌ Start error: {e}") self.stop() return False def _connect(self, timeout: float = 10.0) -> bool: """Establish Redis connection with HuggingFace Spaces resilience.""" try: logger.info(f"🔌 Connecting to Redis: {self.config.url}") # Create Redis connection with container-optimized settings connection_kwargs = { 'decode_responses': self.config.decode_responses, 'socket_timeout': self.config.socket_timeout, 'socket_connect_timeout': self.config.socket_connect_timeout, 'health_check_interval': self.config.health_check_interval, 'retry_on_timeout': self.config.retry_on_timeout, } if self.config.password: connection_kwargs['password'] = self.config.password self._redis = redis.from_url(self.config.url, **connection_kwargs) # Test connection self._redis.ping() # Create pub/sub connection (separate for blocking operations) pubsub_kwargs = connection_kwargs.copy() # ✅ ROOT CAUSE FIX: Was socket_timeout=None which blocks FOREVER # If Redis silently drops the subscription (TCP stays alive), the thread # hangs indefinitely — no error, no reconnect, no recovery. # With 30s timeout, we get periodic TimeoutError which we use as a heartbeat. pubsub_kwargs['socket_timeout'] = 30 # Heartbeat every 30s pubsub_redis = redis.from_url(self.config.url, **pubsub_kwargs) self._pubsub = pubsub_redis.pubsub(ignore_subscribe_messages=True) self._state = ConnectionState.CONNECTED self._connected = True with self._stats_lock: self._stats['connections'] += 1 self._stats['last_connected'] = datetime.now().isoformat() logger.info("✅ Redis connection established") return True except Exception as e: logger.error(f"❌ Redis connection failed: {e}") self._state = ConnectionState.FAILED self._connected = False return False def _start_threads(self): """Start background threads for listening and command processing.""" # Command processing thread self._command_thread = threading.Thread( target=self._command_loop, name="RedisCommandProcessor-V10.1", daemon=True ) self._command_thread.start() # Message listener thread (pub/sub - low latency) self._listener_thread = threading.Thread( target=self._listener_loop, name="RedisListener-V10.1", daemon=True ) self._listener_thread.start() # ✅ Stream catchup thread (Redis Streams - durable recovery) # Inspired by Ably's connection recovery: track timeserials, # resume from last position, recover missed messages on reconnect if self.config.use_streams: self._stream_thread = threading.Thread( target=self._stream_catchup_loop, name="RedisStreamCatchup-V10.2", daemon=True ) self._stream_thread.start() logger.info("🧵 Background threads started (pub/sub + stream catchup)") else: logger.info("🧵 Background threads started") def _command_loop(self): """Process commands from the queue (non-blocking thread).""" while self._running and not self._shutdown_event.is_set(): try: # Get command with timeout to allow periodic shutdown checks command = self._command_queue.get(timeout=1.0) if isinstance(command, StopCommand): break elif isinstance(command, SubscribeCommand): self._handle_subscribe(command) elif isinstance(command, UnsubscribeCommand): self._handle_unsubscribe(command) elif isinstance(command, PublishCommand): self._handle_publish(command) elif isinstance(command, GetStateCommand): self._handle_get_state(command) self._command_queue.task_done() except queue.Empty: continue except Exception as e: logger.error(f"❌ Command processing error: {e}") def _listener_loop(self): """Listen for messages (BLOCKING thread with zero latency). ✅ ROOT CAUSE FIX: Added heartbeat detection via socket_timeout. If no messages arrive for STALL_THRESHOLD seconds, forces full reconnect. This catches silent pub/sub subscription drops that were causing buffer stalls. """ reconnect_attempts = 0 STALL_THRESHOLD = 120 # Force reconnect if no messages in 2 minutes # Track message flow self._last_message_time = time.time() self._listener_alive = True while self._running and not self._shutdown_event.is_set(): try: if not self._pubsub: time.sleep(self.config.reconnect_base_interval) continue # TRUE BLOCKING LISTEN - Zero polling latency! # socket_timeout=30 means this will raise TimeoutError every 30s # if no message arrives, which we use as a heartbeat check. for message in self._pubsub.listen(): if not self._running or self._shutdown_event.is_set(): break if message['type'] == 'message': self._handle_message(message) self._last_message_time = time.time() # ✅ Track last message with self._stats_lock: self._stats['messages_received'] += 1 # Reset reconnect counter on successful operation reconnect_attempts = 0 except (redis.TimeoutError, TimeoutError): # ✅ HEARTBEAT: This fires every ~30s when no messages arrive. # Check if we've gone too long without ANY message. silence_duration = time.time() - self._last_message_time if silence_duration > STALL_THRESHOLD: logger.warning( f"⚠️ [LISTENER] No messages for {silence_duration:.0f}s — " f"forcing reconnect (subscription may have silently dropped)" ) # Force full reconnect self._connected = False self._state = ConnectionState.DISCONNECTED if self._connect(): logger.info("✅ [LISTENER] Reconnected after silent stall") self._resubscribe_all() self._last_message_time = time.time() with self._stats_lock: self._stats['reconnections'] += 1 if self.on_connected: self.on_connected() else: logger.error("❌ [LISTENER] Reconnection failed after silent stall") else: # Normal timeout — subscription is fine, just quiet pass continue # Back to listen() except redis.ConnectionError as e: logger.warning(f"⚠️ Redis connection lost: {e}") self._connected = False self._state = ConnectionState.DISCONNECTED if self.on_disconnected: self.on_disconnected() # Exponential backoff reconnection if reconnect_attempts < 10: # Cap exponential growth reconnect_attempts += 1 delay = min( self.config.reconnect_base_interval * (2 ** reconnect_attempts), self.config.reconnect_max_interval ) logger.info(f"🔄 Reconnecting in {delay:.1f}s... (attempt {reconnect_attempts})") time.sleep(delay) # Attempt reconnection if self._connect(): logger.info("✅ Reconnected successfully") self._resubscribe_all() self._last_message_time = time.time() with self._stats_lock: self._stats['reconnections'] += 1 if self.on_connected: self.on_connected() except Exception as e: logger.error(f"❌ Listener error: {e}") time.sleep(1) self._listener_alive = False def _stream_catchup_loop(self): """ ✅ REDIS STREAMS CATCHUP CONSUMER (V10.2) Architecture inspired by: - Ably: Timeserial tracking, resume from exact position, 2-min recovery window - Redis Streams docs: Consumer groups, XREADGROUP, XACK for exactly-once processing - Arxiv (Borgarelli 2024): Lost rewards in distributed RL are catastrophic since policy learning propagates rewards backward in time This thread runs alongside the pub/sub listener: - Pub/sub: Real-time, zero-latency delivery (primary path) - Streams: Durable recovery for messages missed during disconnections When pub/sub is healthy, stream messages are deduped and skipped. When pub/sub stalls (silent disconnect), stream catchup delivers the missed rewards. """ logger.info("🔄 [STREAM_CATCHUP] Starting Redis Streams recovery consumer") # Wait for subscriptions to be set up time.sleep(5) # Create a dedicated Redis connection for stream reading stream_redis = None catchup_stats = { 'messages_recovered': 0, 'messages_deduped': 0, 'read_errors': 0, 'reconnections': 0, } while self._running and not self._shutdown_event.is_set(): try: # Ensure we have a connection if stream_redis is None: connection_kwargs = { 'decode_responses': True, 'socket_timeout': 10, 'socket_connect_timeout': 5, 'retry_on_timeout': True, } if self.config.password: connection_kwargs['password'] = self.config.password stream_redis = redis.from_url(self.config.url, **connection_kwargs) stream_redis.ping() logger.info("🔄 [STREAM_CATCHUP] Connected to Redis for stream reading") # Ensure consumer groups exist for all subscribed channels with self._subscription_lock: for channel in self._subscriptions: stream_key = f"stream:{channel}" try: stream_redis.xgroup_create( stream_key, self._stream_consumer_group, id='$', mkstream=True ) if stream_key not in self._stream_channels: self._stream_channels.append(stream_key) except redis.ResponseError as e: if "BUSYGROUP" in str(e): if stream_key not in self._stream_channels: self._stream_channels.append(stream_key) else: logger.warning(f"⚠️ [STREAM_CATCHUP] Group create error: {e}") # Build stream read dict: {stream_key: '>'} for new undelivered messages if not self._stream_channels: time.sleep(2) continue streams_to_read = {sk: '>' for sk in self._stream_channels} # XREADGROUP: Block for up to 5 seconds waiting for new messages # This is the Ably-inspired "resume from position" mechanism results = stream_redis.xreadgroup( self._stream_consumer_group, self._stream_consumer_name, streams_to_read, count=50, # Process up to 50 messages per batch block=5000, # 5 second block timeout noack=False # We want to ACK after processing ) if not results: continue for stream_key, messages in results: # Map stream key back to channel name: "stream:rewards" → "rewards" channel_name = stream_key.replace("stream:", "", 1) for msg_id, msg_data in messages: try: # Parse the stream message # Stream data has flat key-value pairs where complex values # were JSON-serialized by the publisher envelope = {} for k, v in msg_data.items(): try: envelope[k] = json.loads(v) except (json.JSONDecodeError, TypeError): envelope[k] = v event_name = envelope.get('event', 'message') data = envelope.get('data', envelope) timestamp = envelope.get('timestamp', time.time()) # ✅ DEDUP: Skip if pub/sub already delivered this message dedup_key = f"{channel_name}:{timestamp}" if dedup_key in self._stream_dedup: catchup_stats['messages_deduped'] += 1 # ACK it — already processed via pub/sub stream_redis.xack(stream_key, self._stream_consumer_group, msg_id) continue # This message was MISSED by pub/sub — deliver it now! redis_msg = RedisMessage( channel=channel_name, name=event_name, data=data, timestamp=timestamp if isinstance(timestamp, float) else time.time() ) # Dispatch to callbacks with self._subscription_lock: if channel_name in self._subscriptions: callback = self._subscriptions[channel_name] if callback: callback(redis_msg) # ACK the message (exactly-once semantics) stream_redis.xack(stream_key, self._stream_consumer_group, msg_id) catchup_stats['messages_recovered'] += 1 if catchup_stats['messages_recovered'] % 100 == 0: logger.info( f"🔄 [STREAM_CATCHUP] Recovered {catchup_stats['messages_recovered']} msgs " f"(deduped: {catchup_stats['messages_deduped']}, " f"errors: {catchup_stats['read_errors']})" ) # Log first recovery prominently if catchup_stats['messages_recovered'] == 1: logger.warning( f"🔄 [STREAM_CATCHUP] *** FIRST RECOVERY *** " f"Delivering missed {channel_name} message via stream! " f"Pub/sub may have dropped this." ) except Exception as e: logger.error(f"❌ [STREAM_CATCHUP] Message processing error: {e}") catchup_stats['read_errors'] += 1 # Still ACK to prevent infinite retry try: stream_redis.xack(stream_key, self._stream_consumer_group, msg_id) except Exception: pass except redis.ConnectionError as e: logger.warning(f"⚠️ [STREAM_CATCHUP] Connection lost: {e}") stream_redis = None catchup_stats['reconnections'] += 1 time.sleep(5) except redis.TimeoutError: # Normal — no new messages in block window continue except Exception as e: logger.error(f"❌ [STREAM_CATCHUP] Error: {e}") catchup_stats['read_errors'] += 1 time.sleep(2) # Cleanup if stream_redis: try: stream_redis.close() except Exception: pass logger.info( f"🔄 [STREAM_CATCHUP] Stopped. Final stats: " f"recovered={catchup_stats['messages_recovered']}, " f"deduped={catchup_stats['messages_deduped']}, " f"errors={catchup_stats['read_errors']}" ) def _handle_message(self, message: dict): """Handle incoming Redis message. For channels subscribed with latest_only=True, this method implements transport-side coalescing (the latest-state latch model): - If no callback is currently in-flight for this channel, fire the callback immediately and mark in_flight[channel] = True. - If a callback IS already in-flight (consumer is busy), store the arriving message in pending_latest[channel], overwriting any previously pending-but-unprocessed message. The overwritten message is intentionally discarded — we only ever want the freshest state. - When the callback finishes, it checks pending_latest and, if a newer message arrived while it was busy, dispatches that single pending message and clears the slot. All intermediate arrivals are silently dropped. Channels NOT registered as latest_only pass straight through the original queue-style path and are completely unaffected. """ try: channel = message['channel'] raw_data = message['data'] # Parse message envelope try: if isinstance(raw_data, str): envelope = json.loads(raw_data) event_name = envelope.get('event', 'message') data = envelope.get('data', envelope) timestamp = envelope.get('timestamp', time.time()) else: event_name = 'message' data = raw_data timestamp = time.time() except (json.JSONDecodeError, TypeError): event_name = 'message' data = raw_data timestamp = time.time() # ✅ DEDUP: Track this message so stream catchup doesn't re-deliver it if self.config.use_streams and isinstance(timestamp, (int, float)): dedup_key = f"{channel}:{timestamp}" self._stream_dedup.add(dedup_key) if len(self._stream_dedup) > self._stream_dedup_max: excess = len(self._stream_dedup) - self._stream_dedup_max // 2 for _ in range(excess): try: self._stream_dedup.pop() except KeyError: break # Create Redis message wrapper redis_msg = RedisMessage( channel=channel, name=event_name, data=data, timestamp=timestamp ) # Call global message handler (always lossless — not coalesced) if self.on_message: self.on_message(channel, redis_msg) # --------------------------------------------------------------- # CHANNEL-SPECIFIC DISPATCH # Separate paths for latest_only (latch) vs normal (queue-style). # --------------------------------------------------------------- with self._subscription_lock: is_latest_only = channel in self._latest_only_channels callback = self._subscriptions.get(channel) if not callback: return if not is_latest_only: # ── LOSSLESS PATH ────────────────────────────────────────── # Standard queue-style dispatch: every message fires the callback # in arrival order. Used for reward, audit, and replay channels. callback(redis_msg) return # ── LATEST-STATE LATCH PATH ──────────────────────────────────── # Redis pub/sub delivers messages in order, but QUASAR must NOT be # forced to walk through stale intermediate feature states one by one. # We implement collapse-to-freshest here inside the manager. with self._latch_lock: busy = self._in_flight.get(channel, False) if busy: # Consumer is still processing the previous message. # Overwrite the pending slot — previous pending is discarded. had_pending = channel in self._pending_latest self._pending_latest[channel] = redis_msg with self._latch_stats_lock: self._latch_stats['busy_drops_avoided'] += 1 if had_pending: # A message that was already waiting gets superseded self._latch_stats['messages_overwritten'] += 1 logger.debug( f"[LATCH] {channel}: consumer busy — pending_latest updated " f"(ts={timestamp:.4f}, overwrite={'yes' if had_pending else 'no'})" ) return # Consumer is idle — claim the in_flight slot and proceed self._in_flight[channel] = True # Fire the callback outside the latch lock to avoid deadlock if the # callback itself calls back into the manager. self._dispatch_latest_and_drain(channel, redis_msg, callback) except Exception as e: logger.error(f"❌ Message handling error: {e}") def _dispatch_latest_and_drain(self, channel: str, redis_msg: Any, callback: Callable): """Fire callback for a latest_only channel and drain any pending slot. This implements the tail of the latch model: 1. Fire callback with the supplied message (marks consumer as busy). 2. After callback returns, atomically check whether a newer message arrived and was stored in pending_latest[channel]. 3. If yes, fire the callback with that newest message and repeat. 4. If no pending message, clear in_flight and return idle. The loop guarantees that if N messages arrive while we are busy, exactly ONE more callback fires (with the freshest of the N), not N callbacks. All intermediate messages are permanently discarded — intentionally. """ while True: try: callback(redis_msg) with self._latch_stats_lock: self._latch_stats['messages_dispatched'] += 1 except Exception as e: logger.error(f"❌ [LATCH] Callback error on {channel}: {e}") # After callback returns, check for a pending-latest message with self._latch_lock: next_msg = self._pending_latest.pop(channel, None) if next_msg is None: # No new message arrived while we were busy — go idle self._in_flight[channel] = False logger.debug(f"[LATCH] {channel}: callback done, no pending — going idle") return # A newer message is waiting; stay in_flight and loop logger.debug( f"[LATCH] {channel}: callback done, dispatching freshest pending " f"(ts={next_msg.timestamp:.4f})" ) redis_msg = next_msg # in_flight stays True — we are about to fire again def _handle_subscribe(self, command: SubscribeCommand): """Handle subscription command. If command.latest_only is True, the channel is registered in the latch model and will receive only the freshest pending message after each processing cycle (collapse-to-freshest). All other channels remain lossless queue-style consumers. """ try: with self._subscription_lock: self._subscriptions[command.channel] = command.callback # ── LATEST-STATE LATCH REGISTRATION ─────────────────────────── # Mark this channel for transport-side coalescing. Initialise # in_flight to False so the first arriving message fires immediately. if command.latest_only: with self._latch_lock: self._latest_only_channels.add(command.channel) self._in_flight.setdefault(command.channel, False) # Clear any stale pending slot from a previous subscription self._pending_latest.pop(command.channel, None) logger.info( f"📡 [LATCH] Subscribed to channel (latest_only=True): {command.channel}" ) else: logger.info(f"📡 Subscribed to channel: {command.channel}") if self._pubsub: self._pubsub.subscribe(command.channel) # ✅ STREAM CATCHUP: Also set up consumer group for this channel's stream if self.config.use_streams and self._redis: stream_key = f"stream:{command.channel}" try: # Create consumer group starting from latest (we don't want historical flood) # On reconnect, we'll read from where we left off self._redis.xgroup_create( stream_key, self._stream_consumer_group, id='$', # Start from latest mkstream=True ) self._stream_channels.append(stream_key) logger.info(f"📡 Stream consumer group created: {stream_key} → {self._stream_consumer_group}") except redis.ResponseError as e: if "BUSYGROUP" in str(e): # Group already exists — that's fine if stream_key not in self._stream_channels: self._stream_channels.append(stream_key) logger.debug(f"📡 Stream consumer group already exists: {stream_key}") else: logger.error(f"❌ Stream group create error for {stream_key}: {e}") except Exception as e: logger.error(f"❌ Stream setup error for {stream_key}: {e}") except Exception as e: logger.error(f"❌ Subscribe error for {command.channel}: {e}") def _handle_unsubscribe(self, command: UnsubscribeCommand): """Handle unsubscription command.""" try: with self._subscription_lock: if command.channel in self._subscriptions: del self._subscriptions[command.channel] # ── LATEST-STATE LATCH CLEANUP ───────────────────────────────── # Remove channel from latch tracking so no stale state leaks if # the channel is later re-subscribed without latest_only. with self._latch_lock: self._latest_only_channels.discard(command.channel) self._in_flight.pop(command.channel, None) self._pending_latest.pop(command.channel, None) if self._pubsub: self._pubsub.unsubscribe(command.channel) logger.info(f"📡 Unsubscribed from channel: {command.channel}") except Exception as e: logger.error(f"❌ Unsubscribe error for {command.channel}: {e}") def _handle_publish(self, command: PublishCommand): """Handle publish command.""" try: if not self._redis: raise ConnectionError("Redis not connected") # Create message envelope envelope = { 'event': command.event_name, 'data': command.message, 'timestamp': time.time() } # Publish message result = self._redis.publish(command.channel, json.dumps(envelope)) with self._stats_lock: self._stats['messages_published'] += 1 # Set result for synchronous callers if command.response_event: command.result = result command.response_event.set() except Exception as e: logger.error(f"❌ Publish error for {command.channel}: {e}") with self._stats_lock: self._stats['publish_errors'] += 1 if command.response_event: command.result = 0 command.response_event.set() def _handle_get_state(self, command: GetStateCommand): """Handle get state command.""" try: with self._subscription_lock: subscriptions = list(self._subscriptions.keys()) with self._stats_lock: stats = self._stats.copy() stats['total_uptime'] = time.time() - self._start_time # ── LATEST-STATE LATCH DIAGNOSTICS ──────────────────────────── with self._latch_lock: latest_only_channels = sorted(self._latest_only_channels) pending_occupancy = sum( 1 for ch in self._latest_only_channels if ch in self._pending_latest ) in_flight_channels = [ch for ch, busy in self._in_flight.items() if busy] with self._latch_stats_lock: latch_stats = self._latch_stats.copy() latch_stats['pending_slot_occupancy'] = pending_occupancy command.result = { 'connected': self._connected, 'state': self._state.name.lower(), 'subscriptions': subscriptions, 'stats': stats, 'reconnect_attempts': 0, # Not tracked in this version 'config': { 'redis_url': self.config.url, 'use_streams': self.config.use_streams, 'socket_timeout': self.config.socket_timeout, 'is_hf_spaces': IS_HF_SPACES }, # Transport-side coalescing diagnostics 'latch_model': { 'latest_only_enabled': bool(latest_only_channels), 'latest_only_channels': latest_only_channels, 'in_flight_channels': in_flight_channels, 'messages_overwritten': latch_stats['messages_overwritten'], 'messages_dispatched': latch_stats['messages_dispatched'], 'busy_drops_avoided': latch_stats['busy_drops_avoided'], 'pending_slot_occupancy': latch_stats['pending_slot_occupancy'], } } except Exception as e: logger.error(f"❌ Get state error: {e}") command.result = {'error': str(e)} finally: command.response_event.set() def _resubscribe_all(self): """Resubscribe to all channels after reconnection.""" try: with self._subscription_lock: channels = list(self._subscriptions.keys()) for channel in channels: if self._pubsub: self._pubsub.subscribe(channel) logger.info(f"🔄 Resubscribed to {len(channels)} channels") except Exception as e: logger.error(f"❌ Resubscription error: {e}") # Public API methods def subscribe( self, channel: str, callback: Optional[Callable] = None, event_filter: Optional[str] = None, latest_only: bool = False, ): """Subscribe to a channel. Args: channel: Redis channel name. callback: Callable invoked with a RedisMessage on each delivery. event_filter: Optional event name filter (passed through to SubscribeCommand). latest_only: If True, enables the latest-state latch model for this channel. While the callback is executing, any new arrivals overwrite the single pending slot rather than accumulating a backlog. When the callback finishes, only the freshest pending message is delivered — all intermediate arrivals are discarded. Use this for hot-path live-state/feature channels where QUASAR must never replay stale feature transactions. Do NOT use for reward, audit, or replay channels — those require lossless delivery and remain queue-style consumers. """ command = SubscribeCommand( channel=channel, callback=callback, event_filter=event_filter, latest_only=latest_only, ) self._command_queue.put(command) def unsubscribe(self, channel: str): """Unsubscribe from a channel.""" command = UnsubscribeCommand(channel=channel) self._command_queue.put(command) def publish(self, channel: str, message: dict, event_name: str = "message") -> int: """Publish a message to a channel.""" response_event = threading.Event() command = PublishCommand( channel=channel, message=message, event_name=event_name, response_event=response_event ) self._command_queue.put(command) # Wait for result (with timeout) if response_event.wait(timeout=5.0): return command.result or 0 else: logger.warning(f"⏰ Publish timeout for channel: {channel}") return 0 def get_state(self, timeout: float = 2.0) -> dict: """Get current connection state.""" command = GetStateCommand() self._command_queue.put(command) if command.response_event.wait(timeout=timeout): return command.result or {} else: return {'error': 'State request timeout'} def stop(self): """Stop the connection manager.""" if not self._running: return logger.info("🛑 Stopping DedicatedRedisConnectionManager...") self._running = False self._shutdown_event.set() # Signal command thread to stop self._command_queue.put(StopCommand()) # Close connections try: if self._pubsub: self._pubsub.unsubscribe() self._pubsub.close() except Exception as e: logger.debug(f"PubSub close error: {e}") try: if self._redis: self._redis.close() except Exception as e: logger.debug(f"Redis close error: {e}") # Wait for threads to finish for thread, name in [(self._command_thread, "Command"), (self._listener_thread, "Listener")]: if thread and thread.is_alive(): thread.join(timeout=2.0) if thread.is_alive(): logger.warning(f"⏰ {name} thread did not stop gracefully") self._state = ConnectionState.CLOSED self._connected = False with self._stats_lock: self._stats['disconnections'] += 1 logger.info("✅ DedicatedRedisConnectionManager stopped") @property def connected(self) -> bool: """Check if connected.""" return self._connected @property def stats(self) -> dict: """Get statistics.""" with self._stats_lock: stats = self._stats.copy() stats['total_uptime'] = time.time() - self._start_time return stats @property def latch_stats(self) -> dict: """Snapshot of latest-state latch coalescing counters (thread-safe). Returns a dict with: latest_only_enabled — True if any channel uses latch mode. messages_overwritten — Arrivals that replaced a waiting pending slot. messages_dispatched — Callbacks actually fired on latest_only channels. busy_drops_avoided — Times a new arrival was coalesced (consumer was busy). pending_slot_occupancy — Current count of occupied pending-latest slots. """ with self._latch_lock: pending_occupancy = sum( 1 for ch in self._latest_only_channels if ch in self._pending_latest ) latest_only_enabled = bool(self._latest_only_channels) with self._latch_stats_lock: snap = self._latch_stats.copy() snap['latest_only_enabled'] = latest_only_enabled snap['pending_slot_occupancy'] = pending_occupancy return snap def get_latch_diagnostics(self) -> dict: """Return detailed latch-model state for monitoring and debugging. Exposes all latch internals without going through the command queue, so it is safe to call from any thread (including QUASAR's main loop) without incurring dispatch latency. Fields: latest_only_channels — Names of all channels in latch mode. in_flight_channels — Channels currently executing their callback. pending_channels — Channels with a waiting (not-yet-dispatched) message. messages_overwritten — Total arrivals that overwrote an existing pending slot. messages_dispatched — Total callbacks fired via the latch path. busy_drops_avoided — Total coalesced arrivals (consumer was busy). pending_slot_occupancy — Current number of occupied pending slots. """ with self._latch_lock: latest_only_channels = sorted(self._latest_only_channels) in_flight_channels = [ch for ch, busy in self._in_flight.items() if busy] pending_channels = list(self._pending_latest.keys()) pending_occupancy = len(pending_channels) with self._latch_stats_lock: snap = self._latch_stats.copy() snap['latest_only_enabled'] = bool(latest_only_channels) snap['latest_only_channels'] = latest_only_channels snap['in_flight_channels'] = in_flight_channels snap['pending_channels'] = pending_channels snap['pending_slot_occupancy'] = pending_occupancy return snap # ============================================================================= # ABLY-COMPATIBLE CHANNEL WRAPPER (HuggingFace Spaces Edition) # ============================================================================= class RedisAblyChannel: """ V10.1: HuggingFace Spaces optimized Ably-compatible channel wrapper. """ def __init__(self, name: str, redis_client: redis.Redis, pubsub: redis.client.PubSub, use_streams: bool = False): self.name = name self._redis = redis_client self._pubsub = pubsub self._use_streams = use_streams self._callbacks: Dict[str, List[Callable]] = {} self._subscribed = False if IS_HF_SPACES: logger.debug(f"🤗 RedisAblyChannel {name} optimized for HuggingFace Spaces") async def publish(self, event_name: str, data: Any) -> int: """Publish message to channel.""" try: envelope = { 'event': event_name, 'data': data, 'timestamp': time.time(), 'channel': self.name } if self._use_streams: # Use Redis Streams for persistence stream_key = f"stream:{self.name}" # FIX v10.1.1: Redis XADD only accepts flat str/int/float/bytes # Serialize nested dicts/lists to JSON strings stream_data = { k: json.dumps(v) if isinstance(v, (dict, list)) else str(v) for k, v in envelope.items() } result = self._redis.xadd(stream_key, stream_data, maxlen=10000) logger.debug(f"📤 Published to stream {stream_key}: {result}") # Also publish to pub/sub for real-time subscribers pub_result = self._redis.publish(self.name, json.dumps(envelope)) return pub_result else: # Standard pub/sub result = self._redis.publish(self.name, json.dumps(envelope)) logger.debug(f"📤 Published to channel {self.name}: {result} subscribers") return result except Exception as e: logger.error(f"❌ Publish error on channel {self.name}: {e}") return 0 async def subscribe(self, event_name: str, callback: Callable): """Subscribe to events on this channel.""" if event_name not in self._callbacks: self._callbacks[event_name] = [] self._callbacks[event_name].append(callback) # Subscribe to Redis channel if not already subscribed if not self._subscribed: self._pubsub.subscribe(self.name) self._subscribed = True logger.info(f"📡 Subscribed to Redis channel: {self.name}") logger.debug(f"📡 Added callback for {self.name}:{event_name}") def _dispatch(self, event_name: str, data: Any, timestamp: float = None): """Dispatch message to callbacks (called by listener thread).""" if event_name in self._callbacks: message = RedisMessage( channel=self.name, name=event_name, data=data, timestamp=timestamp or time.time() ) for callback in self._callbacks[event_name]: try: callback(message) except Exception as e: logger.error(f"❌ Callback error on {self.name}:{event_name}: {e}") class RedisAblyChannelCollection: """Collection of Redis channels (HuggingFace Spaces optimized).""" def __init__(self, redis_client: redis.Redis, pubsub: redis.client.PubSub, use_streams: bool = False): self._redis = redis_client self._pubsub = pubsub self._use_streams = use_streams self._channels: Dict[str, RedisAblyChannel] = {} def get(self, channel_name: str) -> RedisAblyChannel: """Get or create a channel.""" if channel_name not in self._channels: self._channels[channel_name] = RedisAblyChannel( channel_name, self._redis, self._pubsub, self._use_streams ) return self._channels[channel_name] class RedisConnectionState: """Ably-compatible connection state object.""" def __init__(self): self.state = 'initialized' self._callbacks: Dict[str, List[Callable]] = {} def on(self, event: str, callback: Callable): """Register event callback.""" if event not in self._callbacks: self._callbacks[event] = [] self._callbacks[event].append(callback) def _trigger(self, event: str, *args): """Trigger event callbacks.""" if event in self._callbacks: for callback in self._callbacks[event]: try: callback(*args) except Exception as e: logger.error(f"❌ Connection event callback error: {e}") # ============================================================================= # SIMPLIFIED REDIS-ABLY CLIENT (HuggingFace Spaces Edition) # ============================================================================= class RedisAblyClient: """ V10.1: Simplified Redis client with Ably-compatible interface. HuggingFace Spaces optimized with environment-aware configuration. """ def __init__( self, redis_url: Optional[str] = None, key: Optional[str] = None, # Ignored (Ably compatibility) password: Optional[str] = None, use_streams: bool = False, database: int = 0, # ✅ ADDED: Database selection ): if key: logger.info("ℹ️ [RedisAblyClient] Ably key ignored (migration compatibility)") # ✅ FIXED: Use redis_config with database support if redis_url is None: if database == REDIS_DB_FEATURES: redis_url = RedisConfig.for_database(REDIS_DB_FEATURES).url elif database == REDIS_DB_REWARDS: redis_url = RedisConfig.for_database(REDIS_DB_REWARDS).url else: redis_url = RedisConfig.for_database(database).url self._redis_url = redis_url # Connection settings optimized for HuggingFace Spaces connection_kwargs = { 'decode_responses': True, 'socket_timeout': 15 if IS_HF_SPACES else 10, 'socket_connect_timeout': 15 if IS_HF_SPACES else 10, 'retry_on_timeout': True, } if password or REDIS_PASSWORD: connection_kwargs['password'] = password or REDIS_PASSWORD self._redis = redis.Redis.from_url(redis_url, **connection_kwargs) # Separate connection for pub/sub # ✅ FIX: socket_timeout=30 instead of None — enables heartbeat detection # Same fix as DedicatedRedisConnectionManager. Without this, # Rewards.py's signal listener can silently die, stopping ALL reward generation. pubsub_kwargs = connection_kwargs.copy() pubsub_kwargs['socket_timeout'] = 30 # Heartbeat every 30s self._sub_redis = redis.Redis.from_url(redis_url, **pubsub_kwargs) self._pubsub = self._sub_redis.pubsub(ignore_subscribe_messages=True) # Ably-compatible interfaces self.channels = RedisAblyChannelCollection( self._redis, self._pubsub, use_streams ) self.connection = RedisConnectionState() # Start BLOCKING listener thread self._running = True self._listener_thread = threading.Thread( target=self._listener_loop, daemon=True, name="RedisAblyClientListener-V10.1" ) self._listener_thread.start() self.connection.state = 'connected' self.connection._trigger('connected') if IS_HF_SPACES: logger.info(f"✅ RedisAblyClient V10.1 (HF Spaces) connected to {redis_url}") else: logger.info(f"✅ RedisAblyClient V10.1 connected to {redis_url}") def _listener_loop(self): """ BLOCKING listener using pubsub.listen() - zero polling latency! HuggingFace Spaces optimized with enhanced error handling. ✅ FIX: Added heartbeat via socket_timeout=30 and stall detection. This is the Rewards.py side — if this listener dies, Rewards.py stops receiving signals and NO rewards get calculated for ANY consumer. """ reconnect_attempts = 0 STALL_THRESHOLD = 120 # Force reconnect after 2 min silence self._last_ably_message_time = time.time() while self._running: try: # TRUE BLOCKING LISTEN (with 30s heartbeat timeout) for message in self._pubsub.listen(): if not self._running: break if message['type'] == 'message': channel_name = message['channel'] raw_data = message['data'] self._last_ably_message_time = time.time() try: envelope = json.loads(raw_data) event_name = envelope.get('event', 'message') data = envelope.get('data', envelope) timestamp = envelope.get('timestamp', time.time()) except (json.JSONDecodeError, TypeError): event_name = 'message' data = raw_data timestamp = time.time() if channel_name in self.channels._channels: self.channels._channels[channel_name]._dispatch( event_name, data, timestamp ) # Reset reconnect counter on successful operation reconnect_attempts = 0 except (redis.TimeoutError, TimeoutError): # ✅ HEARTBEAT: Fires every ~30s when no messages arrive silence = time.time() - self._last_ably_message_time if silence > STALL_THRESHOLD: logger.warning( f"[RedisAblyClient] ⚠️ No messages for {silence:.0f}s — " f"forcing reconnect (subscription may have silently dropped)" ) # Force full reconnect try: connection_kwargs = { 'decode_responses': True, 'socket_timeout': 30, # ✅ FIX: was None 'retry_on_timeout': True, } if REDIS_PASSWORD: connection_kwargs['password'] = REDIS_PASSWORD self._sub_redis = redis.Redis.from_url(self._redis_url, **connection_kwargs) self._pubsub = self._sub_redis.pubsub(ignore_subscribe_messages=True) # Re-subscribe to all channels for ch in self.channels._channels: self._pubsub.subscribe(ch) self._last_ably_message_time = time.time() logger.info(f"[RedisAblyClient] ✅ Reconnected after silent stall, re-subscribed to {len(self.channels._channels)} channels") except Exception as e: logger.error(f"[RedisAblyClient] ❌ Reconnection failed: {e}") time.sleep(5) continue # Back to listen() except redis.ConnectionError: logger.warning("[RedisAblyClient] ⚠️ Connection lost, reconnecting...") # Exponential backoff for HuggingFace Spaces if reconnect_attempts < 10: reconnect_attempts += 1 delay = min(2 ** reconnect_attempts, 60) if IS_HF_SPACES else min(2 ** reconnect_attempts, 30) time.sleep(delay) try: self._sub_redis.ping() except Exception: # Recreate connection connection_kwargs = { 'decode_responses': True, 'socket_timeout': 30, # ✅ FIX: was None 'retry_on_timeout': True, } if REDIS_PASSWORD: connection_kwargs['password'] = REDIS_PASSWORD self._sub_redis = redis.Redis.from_url(self._redis_url, **connection_kwargs) self._pubsub = self._sub_redis.pubsub(ignore_subscribe_messages=True) # Re-subscribe for ch in self.channels._channels: self._pubsub.subscribe(ch) self._last_ably_message_time = time.time() except Exception as e: if self._running: logger.error(f"[RedisAblyClient] ❌ Listener error: {e}") time.sleep(1 if not IS_HF_SPACES else 2) # Slower retry for containers def close(self): """Close connections.""" self._running = False try: self._pubsub.unsubscribe() self._pubsub.close() except Exception: pass try: self._redis.close() self._sub_redis.close() except Exception: pass self.connection.state = 'closed' logger.info("✅ RedisAblyClient closed") # ============================================================================= # DIAGNOSTIC UTILITY (HuggingFace Spaces Edition) # ============================================================================= def diagnose_redis_connection(system, verbose: bool = True) -> Optional[dict]: """ V10.2: Redis diagnostics with HuggingFace Spaces awareness and latch model reporting. Args: system: Object with .ably_manager attribute verbose: Print detailed output Returns: Connection state dict (includes 'latch_model' sub-dict) """ manager = getattr(system, 'ably_manager', None) if manager is None: logger.warning("⚠️ No Redis manager found on system") return None if verbose: print(f"\n{'='*80}") print(f"📡 REDIS CONNECTION DIAGNOSTICS (V10.2 - HuggingFace Spaces)") print(f"{'='*80}") print(f"🤗 HuggingFace Spaces: {IS_HF_SPACES}") print(f"📋 Redis Config Available: {HAS_REDIS_CONFIG}") try: state = manager.get_state(timeout=5 if IS_HF_SPACES else 2) if verbose: print(f" Connected: {state.get('connected', 'unknown')}") print(f" State: {state.get('state', 'unknown')}") print(f" Reconnect attempts: {state.get('reconnect_attempts', 0)}") config = state.get('config', {}) print(f" Redis URL: {config.get('redis_url', 'unknown')}") print(f" Socket timeout: {config.get('socket_timeout', 'unknown')}") stats = state.get('stats', {}) print(f"\n 📊 Statistics:") print(f" Connections: {stats.get('connections', 0)}") print(f" Disconnections: {stats.get('disconnections', 0)}") print(f" Reconnections: {stats.get('reconnections', 0)}") print(f" Messages recv: {stats.get('messages_received', 0)}") print(f" Messages pub: {stats.get('messages_published', 0)}") print(f" Publish errors: {stats.get('publish_errors', 0)}") print(f" Avg latency: {stats.get('avg_latency_ms', 0):.2f}ms") print(f" Last connected: {stats.get('last_connected', 'never')}") print(f" Uptime: {stats.get('total_uptime', 0):.0f}s") # ── LATCH MODEL DIAGNOSTICS ──────────────────────────────────── latch = state.get('latch_model', {}) print(f"\n 🔒 Latest-State Latch Model (V10.2):") print(f" Enabled: {latch.get('latest_only_enabled', False)}") lo_channels = latch.get('latest_only_channels', []) if lo_channels: print(f" Latest-only channels: {', '.join(lo_channels)}") else: print(f" Latest-only channels: (none)") if_channels = latch.get('in_flight_channels', []) print(f" In-flight now: {', '.join(if_channels) if if_channels else '(none)'}") print(f" Pending slot occupancy: {latch.get('pending_slot_occupancy', 0)}") print(f" Messages dispatched: {latch.get('messages_dispatched', 0)}") print(f" Messages overwritten: {latch.get('messages_overwritten', 0)}") print(f" Busy drops avoided: {latch.get('busy_drops_avoided', 0)}") subs = state.get('subscriptions', []) print(f"\n 📌 Subscriptions ({len(subs)}):") for ch in subs[:10]: tag = " [latest_only]" if ch in lo_channels else "" print(f" • {ch}{tag}") if len(subs) > 10: print(f" ... and {len(subs) - 10} more") print(f"{'='*80}\n") return state except Exception as e: if verbose: print(f" ❌ Diagnostic error: {e}") print(f"{'='*80}\n") return {'error': str(e)} # ============================================================================= # FEATURE BATCH GATEWAY # ============================================================================= # # Architecture (Karl's diagram, 2026-04): # # Market → p_now → Features → 8 per-agent publishes → Redis # │ # FeatureBatchGateway # │ # discards waitlist on new tick # fires only when all N agents # share the same tick / price # │ # complete batch # │ # Erasure / QSAP engine # produce signal at p=p_now # # The gateway is the synchronisation point: # • Features.py publishes each agent's features independently to its own # per-agent Redis channel (xs, s, m, l, xl, xxl, 5m, 10m). # • FeatureBatchGateway subscribes to all N agent channels via the # DedicatedRedisConnectionManager (latest_only=True, so stale same-agent # messages are already coalesced at the transport layer). # • Each arriving message updates that agent's slot in the current-tick # accumulator. # • If a new tick_index arrives while the previous tick's batch is still # incomplete, the incomplete batch is DISCARDED ("waitlist discard") and # the accumulator resets — ensuring Quasar only ever sees feature sets # where all agents share the same p_now. # • When all N agents have contributed to the same tick, the gateway fires # on_batch_ready(batch_payload) exactly once. class FeatureBatchGateway: """ Redis-side batch synchronisation gate. Subscribes to N per-agent feature channels via a DedicatedRedisConnectionManager. Accumulates per-tick contributions and fires on_batch_ready only when all N agents have arrived for the same tick_index. Any partial batch for a superseded tick is silently discarded — the downstream consumer (QSAP / Erasure engine) therefore always receives a coherent, same-price, all-agents feature set. Usage ----- gateway = FeatureBatchGateway( manager = redis_manager, agent_names = ['xs','s','m','l','xl','xxl','5m','10m'], channel_prefix = 'V75:', on_batch_ready = lambda batch: asyncio.ensure_future( system.on_tick_batch_message(batch) ), ) gateway.start() """ def __init__( self, manager, # DedicatedRedisConnectionManager agent_names: list, on_batch_ready, # callable(batch_dict) — fired on complete tick channel_prefix: str = '', # e.g. 'V75:' to match namespaced channels n_required: int = None, # agents required; defaults to len(agent_names) stale_timeout_s: float = 10.0, # discard incomplete batch after this many seconds ): self._manager = manager self._agent_names = list(agent_names) self._n_required = n_required or len(agent_names) self._on_batch = on_batch_ready self._prefix = channel_prefix self._stale_timeout = stale_timeout_s # Accumulator state (guarded by _lock) self._lock = threading.Lock() self._current_tick = None # tick_index for in-progress batch self._batch: dict = {} # agent_name → feature payload self._batch_price: float = 0.0 # price from first agent in current tick self._batch_ts: float = 0.0 # monotonic time when batch was opened # Diagnostics self._batches_fired: int = 0 self._partial_discards: int = 0 self._agent_overwrites: int = 0 # same agent sent twice in one tick self._started = False logger.info("[FeatureBatchGateway] Created: agents=%s n_required=%d prefix=%r", self._agent_names, self._n_required, self._prefix) # ── public API ──────────────────────────────────────────────────────────── def start(self) -> None: """Subscribe to all per-agent channels through the manager.""" if self._started: logger.warning("[FeatureBatchGateway] Already started — ignoring duplicate start()") return for agent in self._agent_names: channel = f"{self._prefix}{agent}" self._manager.subscribe( channel = channel, callback = self._make_handler(agent), latest_only = True, # latch model: coalesce same-agent bursts ) logger.info("[FeatureBatchGateway] Subscribed to channel: %s (latest_only=True)", channel) self._started = True logger.info("[FeatureBatchGateway] Started — waiting for features on %d channels", len(self._agent_names)) def stop(self) -> None: """Unsubscribe from all channels.""" for agent in self._agent_names: channel = f"{self._prefix}{agent}" self._manager.unsubscribe(channel) self._started = False logger.info("[FeatureBatchGateway] Stopped") @property def diagnostics(self) -> dict: with self._lock: return { 'batches_fired': self._batches_fired, 'partial_discards': self._partial_discards, 'agent_overwrites': self._agent_overwrites, 'current_tick': self._current_tick, 'agents_in_batch': len(self._batch), 'n_required': self._n_required, } # ── internal ────────────────────────────────────────────────────────────── def _make_handler(self, agent_name: str): """Return a per-agent message handler closed over agent_name.""" def _handler(message): try: self._receive(agent_name, message) except Exception as e: logger.error("[FeatureBatchGateway] Handler error for %s: %s", agent_name, e) return _handler def _receive(self, agent_name: str, message) -> None: """ Process one arriving per-agent feature message. Called by the DedicatedRedisConnectionManager's listener thread — must be fast and must not block. """ # Unpack the Redis envelope data = getattr(message, 'data', None) if data is None: if isinstance(message, dict): data = message else: logger.debug("[FeatureBatchGateway] %s: unrecognised message shape", agent_name) return # Extract tick_index (price epoch key) and raw price tick_index = ( data.get('tick_index') or (data.get('features') or {}).get('tick_index') or (data.get('features') or {}).get('tick_count') ) raw_price = data.get('price') # envelope-level raw Deriv tick features = data.get('features', {}) now_mono = time.monotonic() with self._lock: # ── Stale-batch eviction ────────────────────────────────────── # If the in-progress batch is older than stale_timeout_s, discard it # before deciding whether the incoming tick matches. if (self._current_tick is not None and (now_mono - self._batch_ts) > self._stale_timeout): n_stale = len(self._batch) logger.warning( "[FeatureBatchGateway] Discarding stale partial batch: " "tick=%s agents=%d/%d age=%.1fs", self._current_tick, n_stale, self._n_required, now_mono - self._batch_ts, ) self._partial_discards += 1 self._batch = {} self._current_tick = None # ── New-tick reset (waitlist discard) ───────────────────────── # This is the core of Karl's diagram: # When a new tick_index arrives while a partial batch is open, # the old partial batch is DISCARDED and we start fresh. # This guarantees: every complete batch that reaches the consumer # has ALL agents at the SAME price / tick. if tick_index is not None and tick_index != self._current_tick: if self._batch: logger.info( "[FeatureBatchGateway] New tick %s arrived — discarding partial " "batch for tick %s (%d/%d agents). Waitlist cleared.", tick_index, self._current_tick, len(self._batch), self._n_required, ) self._partial_discards += 1 self._batch = {} self._current_tick = tick_index self._batch_ts = now_mono self._batch_price = float(raw_price) if raw_price else 0.0 # ── Accumulate ──────────────────────────────────────────────── if agent_name in self._batch: self._agent_overwrites += 1 logger.debug("[FeatureBatchGateway] Agent %s overwrote slot in tick %s", agent_name, self._current_tick) self._batch[agent_name] = { 'features': features, 'price': raw_price, 'tick_index': tick_index, } # Update price from latest arrival if previous was missing if self._batch_price == 0.0 and raw_price: try: self._batch_price = float(raw_price) except (ValueError, TypeError): pass n_now = len(self._batch) logger.debug("[FeatureBatchGateway] tick=%s agents=%d/%d (+%s)", self._current_tick, n_now, self._n_required, agent_name) # ── Fire batch when complete ────────────────────────────────── if n_now >= self._n_required: batch_payload = { 'tick_index': self._current_tick, 'price': self._batch_price, 'timestamp': time.strftime('%H:%M:%S'), 'agent_count': n_now, 'agents': { ag: {'features': v['features'], 'price': v['price']} for ag, v in self._batch.items() }, } self._batches_fired += 1 logger.info( "[FeatureBatchGateway] ✅ BATCH COMPLETE tick=%s price=%.4f " "agents=%d (batch #%d)", self._current_tick, self._batch_price, n_now, self._batches_fired, ) # Reset accumulator BEFORE firing so re-entrant calls don't corrupt state self._batch = {} self._current_tick = None self._batch_price = 0.0 # Fire the callback OUTSIDE the lock to avoid deadlock if the # consumer calls back into the manager. batch_to_fire = batch_payload else: batch_to_fire = None # ── Fire outside lock ───────────────────────────────────────────── if batch_to_fire is not None: try: self._on_batch(batch_to_fire) except Exception as e: logger.error("[FeatureBatchGateway] on_batch_ready raised: %s", e) # ============================================================================= # EXAMPLE USAGE (HuggingFace Spaces Compatible) # ============================================================================= if __name__ == "__main__": import sys logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s' ) def on_message(msg: RedisMessage): print(f"📨 Received: {msg}") def on_connected(): print("🟢 Connected!") def on_disconnected(): print("🔴 Disconnected!") # Example: Synchronous usage with HuggingFace Spaces support print(f"\n=== Testing DedicatedRedisConnectionManager V10.2 ===") print(f"🤗 HuggingFace Spaces: {IS_HF_SPACES}") print(f"📋 Redis Config: {'Available' if HAS_REDIS_CONFIG else 'Using defaults'}") print() manager = DedicatedRedisConnectionManager( on_connected=on_connected, on_disconnected=on_disconnected, on_message=lambda ch, msg: print(f"[Global] {ch}: {msg}"), database=0 # Use default database ) timeout = 10 if IS_HF_SPACES else 5 if manager.start(timeout=timeout): print("✅ Manager started successfully") # ── Lossless subscription (rewards / audit channels) ────────────── manager.subscribe("reward-channel", callback=on_message) # ── Latest-only subscription (live feature / bound-state channels) ─ # Messages arriving while the callback executes are coalesced: # only the FRESHEST pending state is delivered after the callback # finishes; all intermediate arrivals are intentionally discarded. manager.subscribe( "features-live", callback=on_message, latest_only=True, # ← enables transport-side latch model ) # Publish (with timing) time.sleep(1) # Wait for subscriptions start = time.time() result = manager.publish( "features-live", {"hello": "world", "platform": "HuggingFace Spaces" if IS_HF_SPACES else "local"}, "feature-update" ) latency = (time.time() - start) * 1000 print(f"📤 Published to {result} subscribers (latency: {latency:.2f}ms)") # Wait for message time.sleep(2) # Get state state = manager.get_state() print(f"\n📊 Final State:") print(f" Connected: {state.get('connected')}") print(f" Messages sent: {state.get('stats', {}).get('messages_published', 0)}") print(f" Messages received: {state.get('stats', {}).get('messages_received', 0)}") # Print latch model diagnostics latch = manager.get_latch_diagnostics() print(f"\n🔒 Latch Model Diagnostics:") print(f" Latest-only channels: {latch.get('latest_only_channels')}") print(f" Messages dispatched: {latch.get('messages_dispatched')}") print(f" Messages overwritten: {latch.get('messages_overwritten')}") print(f" Busy drops avoided: {latch.get('busy_drops_avoided')}") manager.stop() else: print("❌ Failed to start manager") sys.exit(1)