diff --git a/.gitignore b/.gitignore index c990ddb..467efeb 100644 --- a/.gitignore +++ b/.gitignore @@ -12,4 +12,3 @@ detector_worker.log no_frame_debug.log feeder/ -.venv/ diff --git a/app.py b/app.py index 7cd0407..09cb227 100644 --- a/app.py +++ b/app.py @@ -13,13 +13,7 @@ import requests import asyncio import psutil import zipfile -import ssl -import urllib3 -import subprocess -import tempfile from urllib.parse import urlparse -from requests.adapters import HTTPAdapter -from urllib3.util.ssl_ import create_urllib3_context from fastapi import FastAPI, WebSocket, HTTPException from fastapi.websockets import WebSocketDisconnect from fastapi.responses import Response @@ -246,14 +240,16 @@ async def detect(websocket: WebSocket): logger.debug(f"Processing frame for camera {camera_id} with model {stream['modelId']}") start_time = time.time() - # Extract display identifier for pipeline context + # Extract display identifier for session ID lookup subscription_parts = stream["subscriptionIdentifier"].split(';') display_identifier = subscription_parts[0] if subscription_parts else None + session_id = session_ids.get(display_identifier) if display_identifier else None - # Create context for pipeline execution (session_id will be generated by pipeline) + # Create context for pipeline execution pipeline_context = { "camera_id": camera_id, - "display_id": display_identifier + "display_id": display_identifier, + "session_id": session_id } detection_result = run_pipeline(cropped_frame, model_tree, context=pipeline_context) @@ -263,63 +259,57 @@ async def detect(websocket: WebSocket): # Log the raw detection result for debugging logger.debug(f"Raw detection result for camera {camera_id}:\n{json.dumps(detection_result, indent=2, default=str)}") - # Extract session_id from pipeline result (generated during database record creation) - session_id = None - if detection_result and isinstance(detection_result, dict): - # Check if pipeline generated a session_id (happens when Car+Frontal detected together) - if "session_id" in detection_result: - session_id = detection_result["session_id"] - logger.debug(f"Extracted session_id from pipeline result: {session_id}") - - # Process detection result - run_pipeline returns the primary detection directly - if detection_result and isinstance(detection_result, dict) and "class" in detection_result: - highest_confidence_detection = detection_result - else: - # No detection found + # Direct class result (no detections/classifications structure) + if detection_result and isinstance(detection_result, dict) and "class" in detection_result and "confidence" in detection_result: highest_confidence_detection = { + "class": detection_result.get("class", "none"), + "confidence": detection_result.get("confidence", 1.0), + "box": [0, 0, 0, 0] # Empty bounding box for classifications + } + # Handle case when no detections found or result is empty + elif not detection_result or not detection_result.get("detections"): + # Check if we have classification results + if detection_result and detection_result.get("classifications"): + # Get the highest confidence classification + classifications = detection_result.get("classifications", []) + highest_confidence_class = max(classifications, key=lambda x: x.get("confidence", 0)) if classifications else None + + if highest_confidence_class: + highest_confidence_detection = { + "class": highest_confidence_class.get("class", "none"), + "confidence": highest_confidence_class.get("confidence", 1.0), + "box": [0, 0, 0, 0] # Empty bounding box for classifications + } + else: + highest_confidence_detection = { + "class": "none", + "confidence": 1.0, + "box": [0, 0, 0, 0] + } + else: + highest_confidence_detection = { + "class": "none", + "confidence": 1.0, + "box": [0, 0, 0, 0] + } + else: + # Find detection with highest confidence + detections = detection_result.get("detections", []) + highest_confidence_detection = max(detections, key=lambda x: x.get("confidence", 0)) if detections else { "class": "none", "confidence": 1.0, - "bbox": [0, 0, 0, 0], - "branch_results": {} + "box": [0, 0, 0, 0] } - # Convert detection format to match backend expectations exactly as in worker.md section 4.2 - detection_dict = { - "carModel": None, - "carBrand": None, - "carYear": None, - "bodyType": None, - "licensePlateText": None, - "licensePlateConfidence": None - } + # Convert detection format to match protocol - flatten detection attributes + detection_dict = {} - # Extract and process branch results from parallel classification - branch_results = highest_confidence_detection.get("branch_results", {}) - if branch_results: - logger.debug(f"Processing branch results: {branch_results}") - - # Transform branch results into backend-expected detection attributes - for branch_id, branch_data in branch_results.items(): - if isinstance(branch_data, dict): - logger.debug(f"Processing branch {branch_id}: {branch_data}") - - # Map common classification fields to backend-expected names - if "brand" in branch_data: - detection_dict["carBrand"] = branch_data["brand"] - if "body_type" in branch_data: - detection_dict["bodyType"] = branch_data["body_type"] - if "class" in branch_data: - class_name = branch_data["class"] - - # Map based on branch/model type - if "brand" in branch_id.lower(): - detection_dict["carBrand"] = class_name - elif "bodytype" in branch_id.lower() or "body" in branch_id.lower(): - detection_dict["bodyType"] = class_name - - logger.info(f"Detection payload after branch processing: {detection_dict}") - else: - logger.debug("No branch results found in detection result") + # Handle different detection result formats + if isinstance(highest_confidence_detection, dict): + # Copy all fields from the detection result + for key, value in highest_confidence_detection.items(): + if key not in ["box", "id"]: # Skip internal fields + detection_dict[key] = value detection_data = { "type": "imageDetection", @@ -332,14 +322,12 @@ async def detect(websocket: WebSocket): } } - # Add session ID if available (generated by pipeline when Car+Frontal detected) + # Add session ID if available if session_id is not None: detection_data["sessionId"] = session_id - logger.debug(f"Added session_id to WebSocket response: {session_id}") - if highest_confidence_detection.get("class") != "none": - confidence = highest_confidence_detection.get("confidence", 0.0) - logger.info(f"Camera {camera_id}: Detected {highest_confidence_detection['class']} with confidence {confidence:.2f} using model {stream['modelName']}") + if highest_confidence_detection["class"] != "none": + logger.info(f"Camera {camera_id}: Detected {highest_confidence_detection['class']} with confidence {highest_confidence_detection['confidence']:.2f} using model {stream['modelName']}") # Log session ID if available if session_id: @@ -347,7 +335,6 @@ async def detect(websocket: WebSocket): await websocket.send_json(detection_data) logger.debug(f"Sent detection data to client for camera {camera_id}") - logger.debug(f"Sent this detection data: {detection_data}") return persistent_data except Exception as e: logger.error(f"Error in handle_detection for camera {camera_id}: {str(e)}", exc_info=True) @@ -513,199 +500,6 @@ async def detect(websocket: WebSocket): finally: logger.info(f"Snapshot reader thread for camera {camera_id} is exiting") - async def reconcile_subscriptions(desired_subscriptions, websocket): - """ - Declarative reconciliation: Compare desired vs current subscriptions and make changes - """ - logger.info(f"Reconciling subscriptions: {len(desired_subscriptions)} desired") - - with streams_lock: - # Get current subscriptions - current_subscription_ids = set(streams.keys()) - desired_subscription_ids = set(sub["subscriptionIdentifier"] for sub in desired_subscriptions) - - # Find what to add and remove - to_add = desired_subscription_ids - current_subscription_ids - to_remove = current_subscription_ids - desired_subscription_ids - to_check_for_changes = current_subscription_ids & desired_subscription_ids - - logger.info(f"Reconciliation: {len(to_add)} to add, {len(to_remove)} to remove, {len(to_check_for_changes)} to check for changes") - - # Remove subscriptions that are no longer wanted - for subscription_id in to_remove: - await unsubscribe_internal(subscription_id) - - # Check existing subscriptions for parameter changes - for subscription_id in to_check_for_changes: - desired_sub = next(sub for sub in desired_subscriptions if sub["subscriptionIdentifier"] == subscription_id) - current_stream = streams[subscription_id] - - # Check if parameters changed - if has_subscription_changed(desired_sub, current_stream): - logger.info(f"Parameters changed for {subscription_id}, resubscribing") - await unsubscribe_internal(subscription_id) - await subscribe_internal(desired_sub, websocket) - - # Add new subscriptions - for subscription_id in to_add: - desired_sub = next(sub for sub in desired_subscriptions if sub["subscriptionIdentifier"] == subscription_id) - await subscribe_internal(desired_sub, websocket) - - def has_subscription_changed(desired_sub, current_stream): - """Check if subscription parameters have changed""" - return ( - desired_sub.get("rtspUrl") != current_stream.get("rtsp_url") or - desired_sub.get("snapshotUrl") != current_stream.get("snapshot_url") or - desired_sub.get("snapshotInterval") != current_stream.get("snapshot_interval") or - desired_sub.get("cropX1") != current_stream.get("cropX1") or - desired_sub.get("cropY1") != current_stream.get("cropY1") or - desired_sub.get("cropX2") != current_stream.get("cropX2") or - desired_sub.get("cropY2") != current_stream.get("cropY2") or - desired_sub.get("modelId") != current_stream.get("modelId") or - desired_sub.get("modelName") != current_stream.get("modelName") - ) - - async def subscribe_internal(subscription, websocket): - """Internal subscription logic extracted from original subscribe handler""" - subscriptionIdentifier = subscription.get("subscriptionIdentifier") - rtsp_url = subscription.get("rtspUrl") - snapshot_url = subscription.get("snapshotUrl") - snapshot_interval = subscription.get("snapshotInterval") - model_url = subscription.get("modelUrl") - modelId = subscription.get("modelId") - modelName = subscription.get("modelName") - cropX1 = subscription.get("cropX1") - cropY1 = subscription.get("cropY1") - cropX2 = subscription.get("cropX2") - cropY2 = subscription.get("cropY2") - - # Extract camera_id from subscriptionIdentifier - parts = subscriptionIdentifier.split(';') - if len(parts) != 2: - logger.error(f"Invalid subscriptionIdentifier format: {subscriptionIdentifier}") - return - - display_identifier, camera_identifier = parts - camera_id = subscriptionIdentifier - - # Load model if needed - if model_url: - with models_lock: - if (camera_id not in models) or (modelId not in models[camera_id]): - logger.info(f"Loading model from {model_url} for camera {camera_id}, modelId {modelId}") - extraction_dir = os.path.join("models", camera_identifier, str(modelId)) - os.makedirs(extraction_dir, exist_ok=True) - - # Handle model loading (same as original) - parsed = urlparse(model_url) - if parsed.scheme in ("http", "https"): - filename = os.path.basename(parsed.path) or f"model_{modelId}.mpta" - local_mpta = os.path.join(extraction_dir, filename) - local_path = download_mpta(model_url, local_mpta) - if not local_path: - logger.error(f"Failed to download model from {model_url}") - return - model_tree = load_pipeline_from_zip(local_path, extraction_dir) - else: - if not os.path.exists(model_url): - logger.error(f"Model file not found: {model_url}") - return - model_tree = load_pipeline_from_zip(model_url, extraction_dir) - - if model_tree is None: - logger.error(f"Failed to load model {modelId}") - return - - if camera_id not in models: - models[camera_id] = {} - models[camera_id][modelId] = model_tree - - # Create stream (same logic as original) - if camera_id and (rtsp_url or snapshot_url) and len(streams) < max_streams: - camera_url = snapshot_url if snapshot_url else rtsp_url - - # Check if we already have a stream for this camera URL - shared_stream = camera_streams.get(camera_url) - - if shared_stream: - # Reuse existing stream - buffer = shared_stream["buffer"] - stop_event = shared_stream["stop_event"] - thread = shared_stream["thread"] - mode = shared_stream["mode"] - shared_stream["ref_count"] = shared_stream.get("ref_count", 0) + 1 - else: - # Create new stream - buffer = queue.Queue(maxsize=1) - stop_event = threading.Event() - - if snapshot_url and snapshot_interval: - thread = threading.Thread(target=snapshot_reader, args=(camera_id, snapshot_url, snapshot_interval, buffer, stop_event)) - thread.daemon = True - thread.start() - mode = "snapshot" - shared_stream = { - "buffer": buffer, "thread": thread, "stop_event": stop_event, - "mode": mode, "url": snapshot_url, "snapshot_interval": snapshot_interval, "ref_count": 1 - } - camera_streams[camera_url] = shared_stream - elif rtsp_url: - cap = cv2.VideoCapture(rtsp_url) - if not cap.isOpened(): - logger.error(f"Failed to open RTSP stream for camera {camera_id}") - return - thread = threading.Thread(target=frame_reader, args=(camera_id, cap, buffer, stop_event)) - thread.daemon = True - thread.start() - mode = "rtsp" - shared_stream = { - "buffer": buffer, "thread": thread, "stop_event": stop_event, - "mode": mode, "url": rtsp_url, "cap": cap, "ref_count": 1 - } - camera_streams[camera_url] = shared_stream - else: - logger.error(f"No valid URL provided for camera {camera_id}") - return - - # Create stream info - stream_info = { - "buffer": buffer, "thread": thread, "stop_event": stop_event, - "modelId": modelId, "modelName": modelName, "subscriptionIdentifier": subscriptionIdentifier, - "cropX1": cropX1, "cropY1": cropY1, "cropX2": cropX2, "cropY2": cropY2, - "mode": mode, "camera_url": camera_url, "modelUrl": model_url - } - - if mode == "snapshot": - stream_info["snapshot_url"] = snapshot_url - stream_info["snapshot_interval"] = snapshot_interval - elif mode == "rtsp": - stream_info["rtsp_url"] = rtsp_url - stream_info["cap"] = shared_stream["cap"] - - streams[camera_id] = stream_info - subscription_to_camera[camera_id] = camera_url - logger.info(f"Subscribed to camera {camera_id}") - - async def unsubscribe_internal(subscription_id): - """Internal unsubscription logic""" - if subscription_id in streams: - stream = streams.pop(subscription_id) - camera_url = subscription_to_camera.pop(subscription_id, None) - - if camera_url and camera_url in camera_streams: - shared_stream = camera_streams[camera_url] - shared_stream["ref_count"] -= 1 - - if shared_stream["ref_count"] <= 0: - shared_stream["stop_event"].set() - shared_stream["thread"].join() - if "cap" in shared_stream: - shared_stream["cap"].release() - del camera_streams[camera_url] - - latest_frames.pop(subscription_id, None) - logger.info(f"Unsubscribed from camera {subscription_id}") - async def process_streams(): logger.info("Started processing streams") try: @@ -773,10 +567,6 @@ async def detect(websocket: WebSocket): "modelId": stream["modelId"], "modelName": stream["modelName"], "online": True, - # Include all subscription parameters for proper change detection - "rtspUrl": stream.get("rtsp_url"), - "snapshotUrl": stream.get("snapshot_url"), - "snapshotInterval": stream.get("snapshot_interval"), **{k: v for k, v in get_crop_coords(stream).items() if v is not None} } for camera_id, stream in streams.items() @@ -805,44 +595,29 @@ async def detect(websocket: WebSocket): data = json.loads(msg) msg_type = data.get("type") - if msg_type == "setSubscriptionList": - # Declarative approach: Backend sends list of subscriptions this worker should have - desired_subscriptions = data.get("subscriptions", []) - logger.info(f"Received subscription list with {len(desired_subscriptions)} subscriptions") - - await reconcile_subscriptions(desired_subscriptions, websocket) - - elif msg_type == "subscribe": - # Legacy support - convert single subscription to list - payload = data.get("payload", {}) - await reconcile_subscriptions([payload], websocket) - - elif msg_type == "unsubscribe": - # Legacy support - remove subscription + if msg_type == "subscribe": payload = data.get("payload", {}) subscriptionIdentifier = payload.get("subscriptionIdentifier") - # Remove from current subscriptions and reconcile - current_subs = [] - with streams_lock: - for camera_id, stream in streams.items(): - if stream["subscriptionIdentifier"] != subscriptionIdentifier: - # Convert stream back to subscription format - current_subs.append({ - "subscriptionIdentifier": stream["subscriptionIdentifier"], - "rtspUrl": stream.get("rtsp_url"), - "snapshotUrl": stream.get("snapshot_url"), - "snapshotInterval": stream.get("snapshot_interval"), - "modelId": stream["modelId"], - "modelName": stream["modelName"], - "modelUrl": stream.get("modelUrl", ""), - "cropX1": stream.get("cropX1"), - "cropY1": stream.get("cropY1"), - "cropX2": stream.get("cropX2"), - "cropY2": stream.get("cropY2") - }) - await reconcile_subscriptions(current_subs, websocket) + rtsp_url = payload.get("rtspUrl") + snapshot_url = payload.get("snapshotUrl") + snapshot_interval = payload.get("snapshotInterval") + model_url = payload.get("modelUrl") + modelId = payload.get("modelId") + modelName = payload.get("modelName") + cropX1 = payload.get("cropX1") + cropY1 = payload.get("cropY1") + cropX2 = payload.get("cropX2") + cropY2 = payload.get("cropY2") + + # Extract camera_id from subscriptionIdentifier (format: displayIdentifier;cameraIdentifier) + parts = subscriptionIdentifier.split(';') + if len(parts) != 2: + logger.error(f"Invalid subscriptionIdentifier format: {subscriptionIdentifier}") + continue - elif msg_type == "old_subscribe_logic_removed": + display_identifier, camera_identifier = parts + camera_id = subscriptionIdentifier # Use full subscriptionIdentifier as camera_id for mapping + if model_url: with models_lock: if (camera_id not in models) or (modelId not in models[camera_id]): @@ -1038,10 +813,6 @@ async def detect(websocket: WebSocket): "modelId": stream["modelId"], "modelName": stream["modelName"], "online": True, - # Include all subscription parameters for proper change detection - "rtspUrl": stream.get("rtsp_url"), - "snapshotUrl": stream.get("snapshot_url"), - "snapshotInterval": stream.get("snapshot_interval"), **{k: v for k, v in get_crop_coords(stream).items() if v is not None} } for camera_id, stream in streams.items() diff --git a/docs/MasterElection.md b/docs/MasterElection.md deleted file mode 100644 index c5980b8..0000000 --- a/docs/MasterElection.md +++ /dev/null @@ -1,1449 +0,0 @@ -# Master Election Service Specification - Distributed Process Coordination - -## Overview - -The MasterElection service implements a Redis-based distributed leadership election and process coordination system for the CMS backend cluster. This service provides robust master-slave coordination with automatic failover, process registration, and TTL-based cleanup for multi-process backend deployments. - -**Key Architectural Principle**: Redis-based coordination with atomic Lua scripts ensures consistency and prevents split-brain scenarios while providing automatic cleanup through per-entry TTL expiration. - -## Architecture Components - -### Two-Tier Process Coordination - -The system manages two distinct coordination layers: - -1. **Master Election Layer**: Single leader election across all backend processes -2. **Process Registry Layer**: Individual process registration and heartbeat management - -### Leadership Election Pattern - -- **Single Master**: Only one backend process holds master lock at any time -- **Automatic Failover**: Master election triggers immediately when current master fails -- **Heartbeat-Based**: Master must renew lock every 10 seconds or lose leadership -- **Lua Script Atomicity**: All Redis operations use atomic Lua scripts to prevent race conditions -- **Event-Driven Transitions**: Role changes emit events for dependent services integration - -## Core Components - -### MasterElection Class -`cms-backend/services/MasterElection.ts` - -Primary coordination service that handles distributed leadership election and process lifecycle management. - -**Key Responsibilities:** -- Manages master lock acquisition and renewal using atomic Redis operations -- Provides process registration with automatic TTL-based expiration (45 seconds) -- Emits role transition events for dependent service coordination -- Handles slave registration and heartbeat management -- Maintains process-to-channel mapping for message routing - -### Process Management System - -**Process Registration:** -- Each backend process registers with unique UUID-based identifier -- Process metadata includes role, channel name, and capabilities -- TTL-based expiration (45 seconds) with heartbeat renewal -- Automatic cleanup of stale process entries without manual intervention - -**Channel Assignment:** -- Each process gets assigned a unique Redis pub/sub channel -- Channel mapping stored persistently for message routing -- Master process maintains channel-to-process mapping - -## Data Structures - -### MasterElectionEvents -```typescript -interface MasterElectionEvents { - 'master-acquired': () => void; // This process became master - 'master-lost': () => void; // This process lost master status - 'election-started': () => void; // Election process initiated - 'election-completed': (isMaster: boolean) => void; // Election finished - 'slave-registered': (slave: SlaveNode) => void; // New slave joined - 'slave-removed': (nodeId: string) => void; // Slave left/expired - 'error': (error: Error) => void; // Election/coordination errors -} -``` - -### ProcessInfo -```typescript -interface ProcessInfo { - processId: string; // Unique process identifier (UUID) - nodeId: string; // Node identifier (same as processId) - role: 'master' | 'slave'; // Current process role - lastSeen: string; // Last heartbeat timestamp (ISO string) - capabilities: ProcessCapabilities; // Process feature capabilities -} - -// Channel name derived as: `worker:slave:${processInfo.processId}` -``` - -### ProcessCapabilities -```typescript -interface ProcessCapabilities { - canProcessDetections: boolean; // Can handle AI detection processing - maxSubscriptions: number; // Maximum camera subscriptions supported - preferredWorkload: number; // Preferred subscription load (0-100) -} -``` - -### SlaveNode -```typescript -interface SlaveNode { - nodeId: string; // Unique slave node identifier - identifier: string; // Human-readable process identifier - registeredAt: string; // Initial registration timestamp - lastSeen: string; // Last heartbeat timestamp - metadata?: Record; // Optional process metadata -} -``` - -## Redis Data Architecture - -### Master Election Keys -- `master-election:master` - Current master process identifier with TTL lock -- `master-election:heartbeat` - Master heartbeat timestamp for liveness detection -- `master-election:master_process` - Detailed master process information (JSON) - -### Process Registry Keys (TTL-Enabled) -- `master-election:processes` - Hash map of all active processes with per-entry TTL (45s) -- Channel names derived directly from process ID: `worker:slave:{processId}` - no separate mapping needed - -### TTL Configuration -```typescript -// Per-entry TTL using hSetEx for automatic cleanup -PROCESS_TTL = 45; // Process registration expires after 45 seconds -HEARTBEAT_RENEWAL_INTERVAL = 10; // Process heartbeats renew TTL every 10 seconds -MASTER_LOCK_TTL = 30; // Master lock expires after 30 seconds -``` - -### Data Persistence Strategy -Uses **per-entry TTL with hSetEx** for automatic cleanup: -- Process entries automatically expire if heartbeats stop -- No manual cleanup processes required -- Prevents memory leaks from crashed processes -- Self-healing system that maintains only active processes -- Slave information derived from processes with role='slave' - no separate storage needed -- Channel names derived directly from process ID - no mapping table required - -## Master Election Algorithm - -### Election Flow Diagram - -```mermaid -graph TB - subgraph "Election Process" - START[Process Starts] --> ATTEMPT[attemptElection] - ATTEMPT --> ACQUIRE{acquireMasterLock} - - ACQUIRE -->|Success| MASTER[becomeMaster] - ACQUIRE -->|Failed| SLAVE[becomeSlave] - - MASTER --> HEARTBEAT[startHeartbeat] - SLAVE --> REGISTER[registerAsSlave] - - HEARTBEAT --> RENEW{renewMasterLock} - RENEW -->|Success| CONTINUE[Continue as Master] - RENEW -->|Failed| STEPDOWN[Step Down → SLAVE] - - REGISTER --> MONITOR[Monitor Master] - MONITOR --> CHECK{Master Exists?} - CHECK -->|Yes| WAIT[Wait and Monitor] - CHECK -->|No| ATTEMPT - - STEPDOWN --> SLAVE - WAIT --> MONITOR - CONTINUE --> RENEW - end - - subgraph "Atomic Operations" - ACQUIRE --> LUA1[Lua Script: SET master NX + SET heartbeat] - RENEW --> LUA2[Lua Script: Check owner + PEXPIRE + SET heartbeat] - STEPDOWN --> LUA3[Lua Script: Check owner + DEL master + DEL heartbeat] - end -``` - -### Atomic Lock Operations - -#### Master Lock Acquisition -```lua --- Atomic master lock acquisition with heartbeat -if redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2]) then - redis.call("SET", KEYS[2], ARGV[3], "PX", ARGV[2]) - return 1 -else - return 0 -end -``` - -#### Master Lock Renewal -```lua --- Atomic master lock renewal with heartbeat update -if redis.call("GET", KEYS[1]) == ARGV[1] then - redis.call("PEXPIRE", KEYS[1], ARGV[2]) - redis.call("SET", KEYS[2], ARGV[3], "PX", ARGV[2]) - return 1 -else - return 0 -end -``` - -#### Master Lock Release -```lua --- Atomic master lock release -if redis.call("GET", KEYS[1]) == ARGV[1] then - redis.call("DEL", KEYS[1], KEYS[2]) - return 1 -else - return 0 -end -``` - -## Process Lifecycle Management - -### Process Registration Flow - -```mermaid -sequenceDiagram - participant P as Process - participant R as Redis - participant M as Master Process - - Note over P,M: Process Registration with TTL - - P->>+P: Generate UUID processId - P->>+P: Determine role (master/slave) - P->>+P: Assign channel name - - P->>+R: hSetEx(processes, processId, processInfo, {EX: 45}) - R-->>-P: Registration confirmed - - P->>+R: hSet(channels, processId, channelName) - R-->>-P: Channel mapping stored - - alt Process becomes master - P->>+R: set(master_process, processInfo) - R-->>-P: Master process registered - P->>+M: emit('master-acquired') - else Process becomes slave - P->>+R: hSet(slaves, nodeId, slaveInfo) - R-->>-P: Slave registered - P->>+M: emit('slave-registered', slaveInfo) - end - - Note over P,M: Heartbeat Loop (Every 10s) - - loop Every 10 seconds - P->>+P: updateProcessHeartbeat(processId) - P->>+R: hSetEx(processes, processId, updatedInfo, {EX: 45}) - Note over R: TTL renewed for 45 seconds - R-->>-P: Heartbeat recorded - end - - Note over P,M: Automatic Expiration (No heartbeat) - - R->>R: 45 seconds pass without heartbeat - R->>R: Process entry automatically expires - Note over R: No manual cleanup needed -``` - -### Master Election Scenarios - -#### Scenario 1: Initial Startup -```mermaid -sequenceDiagram - participant P1 as Process 1 - participant P2 as Process 2 - participant R as Redis - - Note over P1,R: First Process Startup - - P1->>+P1: attemptElection() - P1->>+R: Lua Script: SET master NX - R-->>-P1: Success (no existing master) - - P1->>+P1: becomeMaster() - P1->>+P1: emit('master-acquired') - P1->>+P1: startHeartbeat() every 10s - - Note over P1,R: Second Process Startup - - P2->>+P2: attemptElection() - P2->>+R: Lua Script: SET master NX - R-->>-P2: Failed (master exists) - - P2->>+P2: becomeSlave() - P2->>+R: hSet(slaves, nodeId, slaveInfo) - P2->>+P2: emit('election-completed', false) -``` - -#### Scenario 2: Master Failure and Failover -```mermaid -sequenceDiagram - participant P1 as Master Process - participant P2 as Slave Process 1 - participant P3 as Slave Process 2 - participant R as Redis - - Note over P1,R: Normal Operation - - P1->>+R: Heartbeat renewal every 10s - P2->>+P2: Monitor master existence every 5s - P3->>+P3: Monitor master existence every 5s - - Note over P1,R: Master Failure - - P1--XP1: Process crashes/network failure - - Note over R: Master lock expires after 30s - - R->>R: Master lock TTL expires - - Note over P2,R: Slave Detects Missing Master - - P2->>+R: checkMasterExists() Lua Script - R-->>-P2: Master not found or stale - - P2->>+P2: Random delay (0-2s) to reduce collisions - P2->>+R: attemptElection() - Lua Script: SET master NX - R-->>-P2: Success - became new master - - P2->>+P2: becomeMaster() - P2->>+P2: emit('master-acquired') - - Note over P3,R: Other Slave Detects New Master - - P3->>+R: checkMasterExists() - R-->>-P3: New master found - P3->>+P3: Continue as slave - no election needed -``` - -## TTL-Based Cleanup System - -### Per-Entry TTL Implementation - -```typescript -// Process registration with automatic TTL expiration -public async registerProcess(processInfo: ProcessInfo): Promise { - // Set process registration with 45 second TTL per entry - await redisClient.hSetEx( - this.processesKey, - { - [processInfo.processId]: JSON.stringify(processInfo) - }, - { - expiration: { - type: 'EX', - value: 45 // 45 second TTL per process entry - } - } - ); - - // Map process to channel (no TTL - cleaned up manually) - await redisClient.hSet( - this.processChannelsKey, - processInfo.processId, - processInfo.channelName - ); -} - -// Heartbeat renewal extends TTL automatically -public async updateProcessHeartbeat(processId: string): Promise { - const processData = await redisClient.hGet(this.processesKey, processId); - if (processData) { - const processInfo: ProcessInfo = JSON.parse(processData); - processInfo.lastSeen = new Date().toISOString(); - - // Update process and renew TTL on heartbeat (per-entry TTL) - await redisClient.hSetEx( - this.processesKey, - { - [processId]: JSON.stringify(processInfo) - }, - { - expiration: { - type: 'EX', - value: 45 // Renew 45 second TTL for this specific process entry - } - } - ); - } -} -``` - -### Cleanup Behavior - -```mermaid -graph TB - subgraph "TTL Cleanup Process" - REG[Process Registration] --> TTL[45s TTL Set] - TTL --> HB{Heartbeat Within 45s?} - - HB -->|Yes| RENEW[TTL Renewed to 45s] - HB -->|No| EXPIRE[Entry Automatically Expires] - - RENEW --> HB - EXPIRE --> GONE[Process Removed from Redis] - - GONE --> DETECT[Other Processes Detect Absence] - DETECT --> REBALANCE[Automatic Rebalancing] - end - - subgraph "Manual vs TTL Cleanup" - MANUAL[Manual Cleanup Process] - AUTOMATIC[TTL-Based Cleanup] - - MANUAL -.->|"❌ Complex"| ISSUES[Race Conditions
Memory Leaks
Stale Data] - AUTOMATIC -.->|"✅ Simple"| BENEFITS[Self-Healing
No Race Conditions
Guaranteed Cleanup] - end -``` - -## Event System Architecture - -### Event Emission Flow - -```mermaid -graph TD - subgraph "Election Events" - START[Election Started] --> ATTEMPT[Attempt Lock Acquisition] - ATTEMPT --> SUCCESS{Lock Acquired?} - - SUCCESS -->|Yes| MASTER[Become Master] - SUCCESS -->|No| SLAVE[Become Slave] - - MASTER --> MASTER_EVENT[emit('master-acquired')] - SLAVE --> SLAVE_EVENT[emit('election-completed', false)] - - MASTER_EVENT --> HEARTBEAT[Start Heartbeat Loop] - SLAVE_EVENT --> MONITOR[Start Master Monitoring] - end - - subgraph "Heartbeat Events" - HEARTBEAT --> RENEW{Renew Lock?} - RENEW -->|Success| CONTINUE[Continue as Master] - RENEW -->|Failed| LOST[emit('master-lost')] - - LOST --> STEPDOWN[Step Down to Slave] - STEPDOWN --> TRIGGER[Trigger New Election] - CONTINUE --> HEARTBEAT - end - - subgraph "Slave Management Events" - SLAVE_JOIN[New Slave Joins] --> SLAVE_REG[emit('slave-registered')] - SLAVE_TIMEOUT[Slave Heartbeat Timeout] --> SLAVE_REM[emit('slave-removed')] - - SLAVE_REG --> NOTIFY[Notify Dependent Services] - SLAVE_REM --> CLEANUP[Cleanup Assignments] - end -``` - -### Event Handler Integration - -```typescript -// Example: Camera module integration with MasterElection events -const masterElection = getMasterElection(); - -masterElection.on('master-acquired', () => { - // This process became master - start managing workers - masterSlaveWorkerCluster.becomeMaster(); - logger.info('Camera cluster: Became master, connecting to workers'); -}); - -masterElection.on('master-lost', () => { - // This process lost master status - become slave - masterSlaveWorkerCluster.becomeSlave(); - logger.info('Camera cluster: Became slave, disconnecting workers'); -}); - -masterElection.on('slave-registered', (slave: SlaveNode) => { - // New backend process joined - rebalance workload - masterSlaveWorkerCluster.handleSlaveJoined(slave); - logger.info(`Camera cluster: Slave ${slave.nodeId} joined`); -}); - -masterElection.on('slave-removed', (nodeId: string) => { - // Backend process left - reassign its workload - masterSlaveWorkerCluster.handleSlaveLeft(nodeId); - logger.info(`Camera cluster: Slave ${nodeId} removed`); -}); -``` - -## Process Coordination Patterns - -### Master Role Responsibilities - -```mermaid -graph TB - subgraph "Master Process Duties" - LOCK[Maintain Master Lock] --> HEARTBEAT[Send Heartbeats Every 10s] - HEARTBEAT --> MONITOR[Monitor All Slave Processes] - - MONITOR --> CLEANUP[Cleanup Stale Slave Entries] - CLEANUP --> BALANCE[Coordinate Resource Balancing] - - BALANCE --> WORKERS[Manage Worker Connections] - WORKERS --> ROUTE[Route Messages to Slaves] - - ROUTE --> STATUS[Provide Cluster Status] - STATUS --> LOCK - end - - subgraph "Master Failure Scenarios" - NETWORK[Network Partition] --> TIMEOUT[Lock Renewal Timeout] - CRASH[Process Crash] --> TIMEOUT - OVERLOAD[Resource Overload] --> TIMEOUT - - TIMEOUT --> EXPIRE[Master Lock Expires] - EXPIRE --> ELECTION[New Election Triggered] - ELECTION --> RECOVER[New Master Elected] - end -``` - -### Slave Role Responsibilities - -```mermaid -graph TB - subgraph "Slave Process Duties" - REGISTER[Register with Master Election] --> HEARTBEAT[Send Heartbeats Every 5s] - HEARTBEAT --> MONITOR[Monitor Master Existence] - - MONITOR --> PROCESS[Process Assigned Messages] - PROCESS --> REPORT[Report Status to Master] - - REPORT --> DETECT{Master Missing?} - DETECT -->|No| MONITOR - DETECT -->|Yes| ELECTION[Trigger Election] - - ELECTION --> ATTEMPT{Win Election?} - ATTEMPT -->|Yes| PROMOTE[Become Master] - ATTEMPT -->|No| CONTINUE[Continue as Slave] - - PROMOTE --> MASTER[Master Role Duties] - CONTINUE --> REGISTER - end -``` - -## Class Responsibilities Overview - -### Core Class Functions - -| Class | Primary Responsibility | Key Methods | Process Type | -|-------|----------------------|-------------|--------------| -| **MasterElection** | Distributed coordination and leadership election | • `start()` - Initialize election process
• `attemptElection()` - Try to acquire master lock
• `becomeMaster()` - Transition to master role
• `becomeSlave()` - Transition to slave role
• `waitForElectionComplete()` - Synchronous election waiting | Both Master & Slave | -| **Process Registry** | Process lifecycle management | • `registerProcess()` - Register with TTL
• `updateProcessHeartbeat()` - Renew TTL
• `getAllProcesses()` - Get active processes
• `getProcessesByRole()` - Filter by master/slave
• `unregisterProcess()` - Manual cleanup | Both Master & Slave | -| **Master Lock Manager** | Atomic lock operations | • `acquireMasterLock()` - Lua script lock acquisition
• `renewMasterLock()` - Lua script lock renewal
• `releaseMasterLock()` - Lua script lock release
• `checkMasterExists()` - Lua script master validation | Both Master & Slave | -| **Slave Management** | Slave registration and monitoring | • `registerAsSlave()` - Register as slave node
• `updateSlaveHeartbeat()` - Update slave status
• `cleanupStaleSlaves()` - Remove expired slaves
• `getSlaves()` - Get all registered slaves | Both Master & Slave | - -## Object Relationship Diagrams - -### Core Class Structure and Dependencies - -```mermaid -classDiagram - class MasterElection { - -nodeId: string - -identifier: string - -isMaster: boolean - -lockTtl: number - -heartbeatInterval: number - +start() - +stop() - +getIsMaster(): boolean - +getNodeId(): string - +waitForElectionComplete(): Promise~boolean~ - -attemptElection() - -acquireMasterLock(): Promise~boolean~ - -renewMasterLock(): Promise~boolean~ - -releaseMasterLock() - -becomeMaster() - -becomeSlave() - -checkMasterExists(): Promise~boolean~ - } - - class ProcessRegistry { - +registerProcess(processInfo) - +updateProcessHeartbeat(processId) - +getAllProcesses(): Promise~ProcessInfo[]~ - +getMasterProcess(): Promise~ProcessInfo~ - +getProcessesByRole(role): Promise~ProcessInfo[]~ - +unregisterProcess(processId) - +getProcessChannel(processId): Promise~string~ - } - - class SlaveManagement { - +registerAsSlave() - +unregisterFromSlaves() - +updateSlaveHeartbeat() - +getSlaves(): Promise~SlaveNode[]~ - +getSlave(nodeId): Promise~SlaveNode~ - +getSlaveCount(): Promise~number~ - -cleanupStaleSlaves() - -startSlaveManagement() - -stopSlaveManagement() - } - - class EventEmitter { - +on(event, listener) - +emit(event, ...args) - +once(event, listener) - +off(event, listener) - } - - MasterElection --|> EventEmitter : extends - MasterElection --* ProcessRegistry : contains - MasterElection --* SlaveManagement : contains - - MasterElection --> Redis : uses for coordination - ProcessRegistry --> Redis : uses hSetEx for TTL - SlaveManagement --> Redis : uses for slave state -``` - -### Redis Operations and Key Management - -```mermaid -graph TB - subgraph "Redis Key Structure" - MASTER[master-election:master
String - Current master ID with TTL] - HEARTBEAT[master-election:heartbeat
String - Master heartbeat timestamp] - MASTER_PROC[master-election:master_process
String - Master ProcessInfo JSON] - - PROCESSES[master-election:processes
Hash - ProcessInfo with per-entry TTL] - CHANNELS[master-election:channels
Hash - ProcessID → Channel mapping] - SLAVES[master-election:slaves
Hash - SlaveNode data] - end - - subgraph "Atomic Operations" - LUA1[Master Acquisition
SET master NX + SET heartbeat] - LUA2[Master Renewal
Check owner + PEXPIRE + SET heartbeat] - LUA3[Master Release
Check owner + DEL master + heartbeat] - LUA4[Master Check
GET master + GET heartbeat + validate TTL] - end - - subgraph "TTL Operations" - HSETEX1[Process Registration
hSetEx with 45s TTL per entry] - HSETEX2[Heartbeat Renewal
hSetEx renews TTL to 45s] - AUTO[Automatic Expiration
Redis removes expired entries] - end - - MASTER --> LUA1 - MASTER --> LUA2 - MASTER --> LUA3 - HEARTBEAT --> LUA1 - HEARTBEAT --> LUA2 - HEARTBEAT --> LUA4 - - PROCESSES --> HSETEX1 - PROCESSES --> HSETEX2 - PROCESSES --> AUTO -``` - -## Method Call Flow Analysis - -### Election and Role Transition Flow - -```mermaid -sequenceDiagram - participant App as Application - participant ME as MasterElection - participant R as Redis - participant Dep as Dependent Services - - Note over App,Dep: Election Initialization - - App->>+ME: start() - ME->>+ME: attemptElection() - ME->>+ME: emit('election-started') - - ME->>+R: Lua Script: acquireMasterLock() - - alt Lock acquired successfully - R-->>-ME: Success (1) - ME->>+ME: becomeMaster() - ME->>+ME: startHeartbeat() - every 10s - ME->>+ME: startSlaveManagement() - ME->>+Dep: emit('master-acquired') - ME->>+ME: emit('election-completed', true) - else Lock acquisition failed - R-->>-ME: Failed (0) - ME->>+ME: becomeSlave() - ME->>+R: hSet(slaves, nodeId, slaveInfo) - ME->>+ME: startPeriodicCheck() - every 5s - ME->>+Dep: emit('election-completed', false) - end - - Note over App,Dep: Heartbeat and Monitoring Loop - - loop Every 10 seconds (Master) / 5 seconds (Slave) - alt Process is Master - ME->>+R: Lua Script: renewMasterLock() - alt Renewal successful - R-->>-ME: Success (1) - ME->>+ME: Continue as master - else Renewal failed - R-->>-ME: Failed (0) - ME->>+ME: becomeSlave() - ME->>+Dep: emit('master-lost') - ME->>+ME: attemptElection() after delay - end - else Process is Slave - ME->>+R: Lua Script: checkMasterExists() - alt Master exists and healthy - R-->>-ME: Master found (1) - ME->>+ME: Continue monitoring - else No master or stale - R-->>-ME: No master (0) - ME->>+ME: attemptElection() with random delay - end - end - end -``` - -### Process Registration and TTL Management Flow - -```mermaid -sequenceDiagram - participant P as Process - participant ME as MasterElection - participant R as Redis - participant Auto as Redis TTL - - Note over P,Auto: Process Registration with TTL - - P->>+ME: registerProcess(processInfo) - - ME->>+R: hSetEx(processes, processId, processInfo, {EX: 45}) - Note over R: Entry set with 45 second TTL - R-->>-ME: Registration confirmed - - ME->>+R: hSet(channels, processId, channelName) - R-->>-ME: Channel mapping stored - - alt Process is master - ME->>+R: set(master_process, processInfo) - R-->>-ME: Master process info stored - end - - ME-->>-P: Registration complete - - Note over P,Auto: Heartbeat Loop (Every 10s) - - loop Every 10 seconds - P->>+ME: updateProcessHeartbeat(processId) - - ME->>+R: hGet(processes, processId) - R-->>-ME: Current process data - - ME->>+ME: Update lastSeen timestamp - - ME->>+R: hSetEx(processes, processId, updatedInfo, {EX: 45}) - Note over R: TTL renewed to 45 seconds - R-->>-ME: Heartbeat recorded - - ME-->>-P: Heartbeat updated - end - - Note over P,Auto: Automatic TTL Expiration (No heartbeat) - - Note over Auto: 45 seconds pass without heartbeat - Auto->>Auto: Process entry automatically expires - Auto->>R: Remove expired entry from hash - - Note over P,Auto: Other processes detect absence - - P->>+ME: getAllProcesses() - ME->>+R: hGetAll(processes) - R-->>-ME: Only active processes returned - Note over ME: Expired process not included - ME-->>-P: Updated process list -``` - -## System Architecture Diagrams - -### Master Election Cluster Architecture - -```mermaid -graph TB - subgraph "Backend Process Cluster" - M[Master Process
Elected Leader
🏆] - S1[Slave Process 1
Follower] - S2[Slave Process 2
Follower] - S3[Slave Process N
Follower] - end - - subgraph "Redis Coordination Layer" - R[(Redis Server)] - subgraph "Election Keys" - MK[master-election:master
Lock with TTL] - HK[master-election:heartbeat
Timestamp] - end - subgraph "Process Registry (TTL)" - PK[master-election:processes
Hash with per-entry TTL] - CK[master-election:channels
Process→Channel mapping] - end - subgraph "Slave Management" - SK[master-election:slaves
Slave registration data] - end - end - - subgraph "Dependent Services" - CAM[Camera Module
MasterSlaveWorkerCluster] - DS[Display Service
WebSocket Cluster] - OTHER[Other Services
...] - end - - M ===|Master Lock
Heartbeat Every 10s| MK - M ===|Timestamp Update| HK - M ===|TTL Registration
Heartbeat Renewal| PK - - S1 <-->|Monitor Master
Every 5s| R - S2 <-->|Monitor Master
Every 5s| R - S3 <-->|Monitor Master
Every 5s| R - - S1 ===|Slave Registration
Heartbeat Every 5s| SK - S2 ===|Slave Registration
Heartbeat Every 5s| SK - S3 ===|Slave Registration
Heartbeat Every 5s| SK - - M -.->|master-acquired
slave-registered
slave-removed| CAM - M -.->|Role transition events| DS - M -.->|Coordination events| OTHER - - S1 -.->|election-completed
master-lost| CAM - S2 -.->|Election events| DS - S3 -.->|Status events| OTHER -``` - -### TTL-Based Cleanup Architecture - -```mermaid -graph TB - subgraph "Process Lifecycle with TTL" - START[Process Starts] --> REG[Register with 45s TTL] - REG --> ACTIVE[Process Active] - - ACTIVE --> HB{Heartbeat?} - HB -->|Every 10s| RENEW[Renew TTL to 45s] - HB -->|Missed| COUNT[Count down TTL] - - RENEW --> ACTIVE - COUNT --> EXPIRE{TTL = 0?} - EXPIRE -->|No| COUNT - EXPIRE -->|Yes| CLEANUP[Redis Auto-Remove] - - CLEANUP --> DETECT[Other Processes Detect] - DETECT --> REBALANCE[Trigger Rebalancing] - end - - subgraph "Traditional Manual Cleanup vs TTL" - subgraph "❌ Manual Cleanup Problems" - RACE[Race Conditions] - LEAK[Memory Leaks] - STALE[Stale Data] - COMPLEX[Complex Logic] - end - - subgraph "✅ TTL-Based Benefits" - AUTO[Automatic Cleanup] - RELIABLE[Reliable Expiration] - SIMPLE[Simple Implementation] - SELF[Self-Healing] - end - end - - subgraph "TTL Management Operations" - HSETEX[hSetEx(key, field, value, {EX: 45})] - RENEWAL[Heartbeat renews TTL automatically] - EXPIRY[Redis removes expired entries] - - HSETEX --> RENEWAL - RENEWAL --> EXPIRY - EXPIRY --> HSETEX - end -``` - -### Election Timing and Coordination - -```mermaid -gantt - title Master Election Timeline - dateFormat X - axisFormat %s - - section Master Lock - Master Lock TTL (30s) :milestone, m1, 0, 0s - Lock Renewal (10s) :10, 20s - Lock Renewal (10s) :20, 30s - Lock Expires :milestone, m2, 30, 30s - - section Process TTL - Process Registration (45s) :milestone, p1, 0, 0s - Heartbeat Renewal (10s) :10, 20s - Heartbeat Renewal (10s) :20, 30s - Heartbeat Renewal (10s) :30, 40s - Process Expires :milestone, p2, 45, 45s - - section Election Events - Initial Election :milestone, e1, 0, 0s - Slave Monitoring (5s) :5, 10s - Slave Monitoring (5s) :10, 15s - Master Failure Detected :milestone, e2, 30, 30s - New Election Started :32, 35s - New Master Elected :milestone, e3, 35, 35s -``` - -## Event System Architecture - -### Event Flow and Dependencies - -```mermaid -graph TD - subgraph "MasterElection Events" - ES[election-started] --> EA{Election Attempt} - EA -->|Success| MA[master-acquired] - EA -->|Failed| EC[election-completed(false)] - - MA --> HB[Start Heartbeat Loop] - EC --> MON[Start Master Monitoring] - - HB --> RENEW{Heartbeat Success?} - RENEW -->|Success| CONT[Continue as Master] - RENEW -->|Failed| ML[master-lost] - - ML --> STEP[Step Down to Slave] - STEP --> MON - - CONT --> HB - MON --> CHECK{Master Missing?} - CHECK -->|Yes| ES - CHECK -->|No| MON - end - - subgraph "Slave Management Events" - SR[slave-registered] --> UP[Update Assignments] - SREM[slave-removed] --> CLEAN[Cleanup Assignments] - - UP --> NOTIFY[Notify Services] - CLEAN --> REBAL[Rebalance Load] - end - - subgraph "Error Handling Events" - ERR[error] --> LOG[Log Error Details] - LOG --> RECOVER[Attempt Recovery] - RECOVER --> ES - end - - subgraph "External Service Integration" - MA -.->|becomeMaster()| CAMERA[Camera Module] - ML -.->|becomeSlave()| CAMERA - SR -.->|slaveJoined()| CAMERA - SREM -.->|slaveLeft()| CAMERA - - MA -.->|Master role| DISPLAY[Display Service] - ML -.->|Slave role| DISPLAY - - MA -.->|Coordinate| OTHER[Other Services] - ML -.->|Follow| OTHER - end -``` - -### Event Sequence Patterns - -#### Master Failure and Recovery Pattern - -```mermaid -sequenceDiagram - participant M as Master Process - participant S1 as Slave 1 - participant S2 as Slave 2 - participant R as Redis - participant Svc as Dependent Services - - Note over M,Svc: Normal Operation - M->>R: Heartbeat renewal every 10s - S1->>R: Monitor master every 5s - S2->>R: Monitor master every 5s - - Note over M,Svc: Master Failure - M--XM: Process crashes - - Note over R: Master lock expires (30s) - R->>R: Lock TTL expires - - Note over S1,S2: Slaves detect master failure - S1->>R: checkMasterExists() → false - S2->>R: checkMasterExists() → false - - Note over S1,S2: Election race with random delay - S1->>S1: Random delay 1.2s - S2->>S2: Random delay 0.8s - - S2->>R: attemptElection() first - R->>S2: Success - became master - S2->>S2: emit('master-acquired') - S2->>Svc: becomeMaster() event - - S1->>R: attemptElection() second - R->>S1: Failed - master exists - S1->>S1: Continue as slave - - Note over S2,Svc: New master operational - S2->>R: Start heartbeat renewal - Svc->>S2: Acknowledge new master -``` - -## Configuration and Tuning - -### Timing Configuration - -```typescript -// MasterElection constructor parameters -interface MasterElectionConfig { - lockName: string = 'master-election'; // Redis key prefix - lockTtl: number = 30000; // Master lock TTL (30 seconds) - heartbeatInterval: number = 10000; // Master heartbeat interval (10 seconds) - checkInterval: number = 5000; // Slave monitoring interval (5 seconds) - identifier: string = 'cms-backend'; // Human-readable process identifier -} - -// TTL Configuration -const PROCESS_TTL_SECONDS = 45; // Process registration TTL -const SLAVE_TIMEOUT_MS = 15000; // Slave cleanup threshold (3x heartbeat) -const ELECTION_RANDOM_DELAY_MAX = 2000; // Max random delay to prevent collisions -``` - -### Redis Key Structure - -```typescript -// Election and coordination keys -const REDIS_KEYS = { - // Master election coordination - master: `${lockName}:master`, // Current master ID with TTL - heartbeat: `${lockName}:heartbeat`, // Master heartbeat timestamp - masterProcess: `${lockName}:master_process`, // Master ProcessInfo JSON - - // Process registry with TTL - processes: `${lockName}:processes`, // Hash: processId → ProcessInfo (TTL per entry) - channels: `${lockName}:channels`, // Hash: processId → channelName - - // Slave management - slaves: `${lockName}:slaves`, // Hash: nodeId → SlaveNode -}; - -// TTL settings -const TTL_CONFIG = { - masterLock: 30, // seconds - Master lock expiration - processEntry: 45, // seconds - Process registration TTL - heartbeatRenewal: 10, // seconds - How often to renew heartbeats - slaveMonitoring: 5, // seconds - How often slaves check master -}; -``` - -### Performance Characteristics - -#### Scalability Metrics -- **Election Speed**: < 100ms for uncontested election -- **Failover Time**: < 5 seconds from master failure to new election -- **Process Registration**: < 10ms per process registration -- **TTL Cleanup**: Automatic, no performance impact on application - -#### Resource Usage -- **Memory**: O(n) where n = number of backend processes -- **Redis Operations**: Atomic Lua scripts prevent race conditions -- **Network**: Minimal - only heartbeats and election attempts -- **CPU**: Negligible overhead for coordination operations - -#### Reliability Guarantees -- **Split-Brain Prevention**: Atomic Lua scripts ensure single master -- **Automatic Recovery**: TTL-based cleanup handles all failure scenarios -- **Event Consistency**: All role transitions emit events for service coordination -- **State Persistence**: Process registry survives Redis restarts - -## Public Interface Specification - -The MasterElection service provides a clean, event-driven interface for distributed coordination across backend processes. - -### Primary Interface: MasterElection Class - -#### Core Lifecycle Methods - -```typescript -/** - * Initialize and start the master election process - * @returns Promise - Resolves when election completes - */ -public async start(): Promise - -/** - * Stop master election and cleanup resources - * @returns Promise - Resolves when cleanup completes - */ -public async stop(): Promise - -/** - * Wait for election to complete with timeout - * @param timeoutMs - Maximum time to wait (default: 30000) - * @returns Promise - true if became master, false if slave - */ -public async waitForElectionComplete(timeoutMs: number = 30000): Promise -``` - -#### Status and Information Methods - -```typescript -/** - * Check if this process is currently the master - * @returns boolean - true if master, false if slave - */ -public getIsMaster(): boolean - -/** - * Get this process's unique node identifier - * @returns string - UUID-based node identifier - */ -public getNodeId(): string - -/** - * Get this process's human-readable identifier - * @returns string - Process identifier (e.g., 'cms-backend') - */ -public getIdentifier(): string - -/** - * Get or set process metadata for coordination - * @param metadata - Optional metadata to set - * @returns Record - Current metadata - */ -public setMetadata(metadata: Record): void -public getMetadata(): Record -``` - -#### Process Registry Methods - -```typescript -/** - * Register a process in the distributed registry with TTL - * @param processInfo - Process information including role and capabilities - * @returns Promise - */ -public async registerProcess(processInfo: ProcessInfo): Promise - -/** - * Update process heartbeat to renew TTL (45 seconds) - * @param processId - Process identifier to update - * @returns Promise - */ -public async updateProcessHeartbeat(processId: string): Promise - -/** - * Get all currently registered processes (auto-filtered by TTL) - * @returns Promise - Array of active processes - */ -public async getAllProcesses(): Promise - -/** - * Get current master process information - * @returns Promise - Master process or null if none - */ -public async getMasterProcess(): Promise - -/** - * Get processes filtered by role - * @param role - 'master' or 'slave' - * @returns Promise - Processes with specified role - */ -public async getProcessesByRole(role: 'master' | 'slave'): Promise -``` - -#### Slave Management Methods - -```typescript -/** - * Get all registered slave nodes - * @returns Promise - Array of active slaves - */ -public async getSlaves(): Promise - -/** - * Get specific slave node information - * @param nodeId - Slave node identifier - * @returns Promise - Slave info or null if not found - */ -public async getSlave(nodeId: string): Promise - -/** - * Get count of registered slave nodes - * @returns Promise - Number of active slaves - */ -public async getSlaveCount(): Promise -``` - -### Event System Interface - -#### Event Registration - -```typescript -// Type-safe event registration -masterElection.on('master-acquired', () => { - // This process became the master - console.log('Became master - start coordinating resources'); -}); - -masterElection.on('master-lost', () => { - // This process lost master status - console.log('Lost master status - step down to slave role'); -}); - -masterElection.on('election-completed', (isMaster: boolean) => { - // Election finished - role determined - console.log(`Election completed - role: ${isMaster ? 'MASTER' : 'SLAVE'}`); -}); - -masterElection.on('slave-registered', (slave: SlaveNode) => { - // New backend process joined cluster - console.log(`New slave joined: ${slave.nodeId}`); -}); - -masterElection.on('slave-removed', (nodeId: string) => { - // Backend process left cluster (TTL expired) - console.log(`Slave removed: ${nodeId}`); -}); - -masterElection.on('error', (error: Error) => { - // Election or coordination error occurred - console.error('Master election error:', error); -}); -``` - -#### Event Timing Guarantees - -- **master-acquired**: Emitted immediately after successful lock acquisition -- **master-lost**: Emitted immediately after failed lock renewal -- **election-completed**: Emitted after initial election resolves (master or slave) -- **slave-registered**: Emitted when new slave joins (master only) -- **slave-removed**: Emitted when slave TTL expires (master only) -- **error**: Emitted on Redis connection issues or election failures - -### Usage Patterns - -#### Basic Initialization and Coordination - -```typescript -import { initialize, getMasterElection } from '~/services/MasterElection'; - -// Initialize master election with custom settings -await initialize( - 'cms-cluster', // lockName - Redis key prefix - 30000, // lockTtl - Master lock TTL (30s) - 10000, // heartbeatInterval - Master heartbeat (10s) - 5000, // checkInterval - Slave monitoring (5s) - 'cms-backend-prod' // identifier - Human-readable name -); - -// Get election instance for event handling -const masterElection = getMasterElection(); - -// Wait for initial election to complete -const isMaster = await masterElection.waitForElectionComplete(); -console.log(`Process started as: ${isMaster ? 'MASTER' : 'SLAVE'}`); -``` - -#### Service Integration Pattern - -```typescript -// Camera module integration example -class CameraClusterService { - private masterElection: MasterElection; - - constructor() { - this.masterElection = getMasterElection(); - this.setupElectionHandlers(); - } - - private setupElectionHandlers() { - // Handle master role transitions - this.masterElection.on('master-acquired', () => { - this.becomeMaster(); - }); - - this.masterElection.on('master-lost', () => { - this.becomeSlave(); - }); - - // Handle cluster membership changes - this.masterElection.on('slave-registered', (slave) => { - this.handleSlaveJoined(slave); - }); - - this.masterElection.on('slave-removed', (nodeId) => { - this.handleSlaveLeft(nodeId); - }); - } - - private async becomeMaster() { - console.log('Camera service: Becoming master'); - - // Connect to all Python ML workers - await this.connectToAllWorkers(); - - // Start managing cluster assignments - this.startClusterManagement(); - - // Begin rebalancing subscriptions - this.startRebalancing(); - } - - private async becomeSlave() { - console.log('Camera service: Becoming slave'); - - // Disconnect from Python workers (master-only) - await this.disconnectFromWorkers(); - - // Stop cluster management - this.stopClusterManagement(); - - // Start listening for routed messages - this.startSlaveMessageHandling(); - } -} -``` - -#### Process Registration with Custom Capabilities - -```typescript -// Register this process with specific capabilities -await masterElection.registerProcess({ - processId: masterElection.getNodeId(), - nodeId: masterElection.getNodeId(), - role: masterElection.getIsMaster() ? 'master' : 'slave', - channelName: `worker:slave:${masterElection.getNodeId()}`, - lastSeen: new Date().toISOString(), - capabilities: { - canProcessDetections: true, // Can handle AI detection callbacks - maxSubscriptions: 100, // Maximum camera subscriptions - preferredWorkload: 80 // Preferred load percentage (0-100) - } -}); - -// Start heartbeat loop to maintain registration -setInterval(async () => { - await masterElection.updateProcessHeartbeat(masterElection.getNodeId()); -}, 10000); // Every 10 seconds -``` - -#### Cluster Monitoring and Status - -```typescript -// Monitor cluster status and health -async function monitorClusterHealth() { - // Get all active processes (TTL-filtered automatically) - const allProcesses = await masterElection.getAllProcesses(); - console.log(`Active processes: ${allProcesses.length}`); - - // Get current master - const masterProcess = await masterElection.getMasterProcess(); - if (masterProcess) { - console.log(`Master: ${masterProcess.processId} (${masterProcess.capabilities.maxSubscriptions} max subscriptions)`); - } - - // Get all slaves - const slaves = await masterElection.getSlaves(); - console.log(`Slaves: ${slaves.length}`); - slaves.forEach(slave => { - console.log(` Slave ${slave.nodeId}: last seen ${slave.lastSeen}`); - }); - - // Check if this process is master - if (masterElection.getIsMaster()) { - console.log('This process is the master - coordinating cluster'); - } else { - console.log('This process is a slave - following master'); - } -} - -// Run monitoring every 30 seconds -setInterval(monitorClusterHealth, 30000); -``` - -#### Graceful Shutdown Pattern - -```typescript -// Graceful shutdown with proper cleanup -process.on('SIGTERM', async () => { - console.log('Shutting down master election...'); - - try { - // Stop election and cleanup resources - await masterElection.stop(); - - // Master automatically releases lock - // Process TTL will expire naturally - // Slaves will detect and trigger new election - - console.log('Master election shutdown complete'); - } catch (error) { - console.error('Error during election shutdown:', error); - } - - process.exit(0); -}); -``` - -### Error Handling and Recovery - -#### Election Failure Scenarios - -```typescript -// Handle various failure modes -masterElection.on('error', (error) => { - console.error('Master election error:', error.message); - - // Common error types: - if (error.message.includes('Redis')) { - // Redis connection issues - console.log('Redis connectivity problem - will retry automatically'); - - } else if (error.message.includes('timeout')) { - // Election timeout - console.log('Election timeout - may indicate network issues'); - - } else if (error.message.includes('lock')) { - // Lock acquisition issues - console.log('Lock contention - normal during elections'); - } - - // Service continues running - election will retry automatically -}); - -// Handle network partitions -masterElection.on('master-lost', () => { - console.log('Lost master status - likely network partition or overload'); - - // Dependent services should gracefully step down - // New election will start automatically after random delay -}); -``` - -#### Recovery Guarantees - -- **Split-Brain Prevention**: Atomic Lua scripts ensure only one master exists -- **Automatic Failover**: New elections triggered immediately when master fails -- **TTL-Based Cleanup**: Processes automatically removed when heartbeats stop -- **State Recovery**: Process registry rebuilds automatically from active heartbeats -- **Event Consistency**: All role changes emit events for service coordination - -### Integration with Dependent Services - -The MasterElection service is designed to coordinate multiple backend services that need distributed leadership: - -#### Camera Module Integration -- Master: Connects to Python ML workers, manages subscriptions -- Slaves: Process routed detection messages, forward commands - -#### Display WebSocket Cluster -- Master: Manages WebSocket connection assignments across processes -- Slaves: Handle assigned display connections, route messages - -#### Database Migration Coordination -- Master: Executes database migrations and schema changes -- Slaves: Wait for master to complete before proceeding - -This specification provides a comprehensive understanding of the MasterElection service's distributed coordination capabilities and integration patterns for multi-process backend systems. \ No newline at end of file diff --git a/docs/WorkerConnection.md b/docs/WorkerConnection.md deleted file mode 100644 index b11ba61..0000000 --- a/docs/WorkerConnection.md +++ /dev/null @@ -1,1498 +0,0 @@ -# Worker Connection Architecture Specification - Pure Declarative State Management - -## Overview - -The Camera Module implements a pure VMware DRS-like declarative architecture for managing connections to Python ML workers. This system uses the database as the single source of truth for desired subscription state, with automatic regeneration and reconciliation providing intelligent camera management, real-time object detection, and AI-powered content selection with automatic load balancing capabilities. - -**Key Architectural Principle**: Database mutations trigger complete state regeneration rather than incremental updates, ensuring consistency and eliminating complex state synchronization issues. - -## Architecture Components - -### Two-Cluster System - -The system consists of two distinct but coordinated clusters: - -1. **Backend Process Cluster**: Multiple CMS backend processes with leader election -2. **Worker Cluster**: Python ML workers for object detection processing - -### Master-Slave WebSocket Architecture - -- **Master Process**: Single elected backend process that maintains WebSocket connections to Python workers -- **Slave Processes**: All other backend processes that handle message routing and processing -- **Message Routing**: Master forwards worker messages to assigned slaves via Redis pub/sub channels -- **MasterElection Integration**: Automated master/slave role management with event-driven transitions -- **Seamless Scaling**: Backend processes can be added/removed without affecting WebSocket connections - -## Core Components - -### DetectorCluster -`cms-backend/modules/camera/services/DetectorCluster.ts` - -Primary interface for camera operations that abstracts the underlying distributed architecture. - -**Key Responsibilities:** -- Routes camera subscription requests through the cluster -- Manages detection callback registration and event emission -- Bridges CameraService with underlying MasterSlaveWorkerCluster -- Provides unified API regardless of master/slave status - -### MasterSlaveWorkerCluster -`cms-backend/modules/camera/services/MasterSlaveWorkerCluster.ts` - -Core distributed cluster implementation that handles declarative state management and worker assignment reconciliation. - -**Master Mode Responsibilities:** -- Maintains WebSocket connections to all Python workers -- Manages desired vs actual subscription state separation -- Implements VMware DRS-like global rebalancing algorithm -- Processes automatic reconciliation every 30 seconds -- Responds to slave join/leave events from MasterElection -- Generates fresh pre-signed model URLs for worker assignments - -**Slave Mode Responsibilities:** -- Submits desired subscription state changes to master -- Processes detection results routed from master -- Event-driven role transitions managed by MasterElection -- No direct worker management (delegated to master) - -### DetectorConnection -`cms-backend/modules/camera/services/DetectorConnection.ts` - -Individual WebSocket connection handler for Python workers. - -**Key Features:** -- Connection lifecycle management (connect, disconnect, reconnect) -- Exponential backoff reconnection with 10-second intervals -- Subscription state management and restoration after reconnection -- Real-time heartbeat monitoring with 10-second timeout -- Resource usage tracking (CPU, memory, GPU) - -## Data Structures - -### WorkerConnectionState -```typescript -interface WorkerConnectionState { - url: string; // Worker WebSocket URL - processId: string; // Backend process managing this worker - online: boolean; // Connection status - cpuUsage: number | null; // Worker CPU utilization - memoryUsage: number | null; // Worker memory usage - gpuUsage: number | null; // Worker GPU utilization - gpuMemoryUsage: number | null; // Worker GPU memory usage - subscriptionCount: number; // Active camera subscriptions - subscriptions: string[]; // List of subscription identifiers - lastHeartbeat: string; // Last heartbeat timestamp - connectedAt: string; // Connection established timestamp -} -``` - -### DesiredCameraSubscription -```typescript -interface DesiredCameraSubscription { - subscriptionIdentifier: string; // Format: ${displayId};${cameraId} - rtspUrl: string; // Camera RTSP stream URL - modelId: number; // AI model database ID - modelName: string; // AI model identifier - createdAt: string; // Subscription creation timestamp - - // Snapshot configuration - snapshotUrl?: string; // Optional snapshot endpoint URL - snapshotInterval?: number; // Snapshot interval in milliseconds - - // Image cropping parameters - cropX1?: number; // Crop region top-left X - cropY1?: number; // Crop region top-left Y - cropX2?: number; // Crop region bottom-right X - cropY2?: number; // Crop region bottom-right Y -} -``` - -### ActualCameraSubscription -```typescript -interface ActualCameraSubscription { - subscriptionIdentifier: string; // Format: ${displayId};${cameraId} - assignedWorkerUrl: string; // Worker handling this subscription - modelUrl: string; // AI model presigned URL (1hr TTL) - status: 'active' | 'pending' | 'failed' | 'recovering'; - assignedAt: string; // Worker assignment timestamp - lastSeen: string; // Last activity timestamp -} -``` - -### SlaveState -```typescript -interface SlaveState { - slaveId: string; // Unique slave identifier (process ID) - processId: string; // Backend process ID (same as slaveId) - online: boolean; // Always true (maintained by MasterElection) - workload: number; // Number of assigned workers (calculated) - lastSeen: string; // Last heartbeat from MasterElection - capabilities?: Record; // Metadata from MasterElection -} -``` - -### DetectorWorkerCommand -```typescript -interface DetectorWorkerCommand { - type: DetectorWorkerCommandType; - payload?: { - subscriptionIdentifier: string; - rtspUrl: string; - snapshotUrl?: string; - snapshotInterval?: number; - modelUrl: string; - modelName: string; - modelId: number; - cropX1?: number; - cropY1?: number; - cropX2?: number; - cropY2?: number; - }; -} - -enum DetectorWorkerCommandType { - SUBSCRIBE = "subscribe", - UNSUBSCRIBE = "unsubscribe", - REQUEST_STATE = "requestState", - PATCH_SESSION_RESULT = "patchSessionResult", - SET_SESSION_ID = "setSessionId" -} -``` - -### ImageDetectionResponse -```typescript -interface ImageDetectionResponse { - subscriptionIdentifier: string; - timestamp: Date; - data: { - detection: { - carModel?: string; - carBrand?: string; - carYear?: number; - bodyType?: string; - licensePlateText?: string; - licensePlateType?: string; - }; - modelId: number; - modelName: string; - }; -} -``` - -## Redis Data Architecture - -### Persistent Storage Keys -- `worker:connections` - Worker connection states and health metrics -- `worker:assignments` - Worker-to-slave assignment mappings -- `worker:desired_subscriptions` - Desired camera subscription state (user intent) -- `worker:actual_subscriptions` - Actual worker subscription assignments (system state) -- `master-election:slaves` - Slave registration and heartbeat (managed by MasterElection) - -### Communication Channels -- `worker:slave:{slaveId}` - Individual slave message routing channels -- `worker:messages:upstream` - Worker-to-master communication channel (currently unused) -- `worker:assignments:changed` - Assignment change broadcast notifications -- `worker:master:commands` - Database change notification channel (slaves → master) - -### Data Persistence Strategy -All Redis data uses **manual cleanup only** (no TTL) to ensure: -- Reliable state recovery after process restarts -- Consistent subscription persistence across failovers -- Predictable cleanup during planned maintenance -- Debug visibility into system state history - -## Pure Declarative Architecture (VMware DRS-like) - -### Concept Overview -The system implements a pure declarative approach similar to VMware Distributed Resource Scheduler (DRS), where: -- **Database**: Single source of truth for desired state (Display+Camera+Playlist combinations) -- **Actual State**: What subscriptions are currently running on workers (stored in `worker:actual_subscriptions`) -- **Regeneration**: Master regenerates complete desired state from database on every change notification -- **Reconciliation**: Master continuously reconciles desired vs actual state via global rebalancing - -### Pure Declarative Benefits -- **Database as Truth**: Desired state always derived fresh from database, eliminating state synchronization issues -- **Zero Incremental Updates**: No complex state management, just "regenerate everything on change" -- **Automatic Recovery**: System heals itself by comparing database state vs actual worker state -- **Load Balancing**: Global optimization across all workers and subscriptions -- **Fault Tolerance**: Desired state survives all failures since it's always derived from database -- **Simplicity**: Database mutations just trigger regeneration - no complex command protocols - -### Pure Declarative Flow -```typescript -// Triggered by any database change -async handleDatabaseChange(changeType: string, entityId: string) { - // 1. Any process detects database change - await triggerSubscriptionUpdate(changeType, entityId); - - // 2. Master receives regeneration request - async handleMasterCommand(message) { - if (data.type === 'regenerate_subscriptions') { - await regenerateDesiredStateFromDatabase(); - } - } - - // 3. Master regenerates complete desired state from database - async regenerateDesiredStateFromDatabase() { - const activeDisplays = await db.display.findMany({ - where: { - AND: [ - { cameraIdentifier: { not: null } }, - { playlistId: { not: null } } - ] - }, - include: { camera: true, playlist: { include: { model: true } } } - }); - - // Generate fresh desired subscriptions from database - await storeDesiredSubscriptions(generateFromDisplays(activeDisplays)); - - // Trigger reconciliation - await rebalanceCameraSubscriptions(); - } - - // 4. Reconciliation (same VMware DRS algorithm) - async rebalanceCameraSubscriptions() { - const desired = await getDesiredSubscriptions(); // Fresh from database - const actual = await getActualSubscriptions(); // Current worker state - - // Find and fix differences using load balancing - await reconcileDifferences(desired, actual); - } -} - -// VMware DRS-like worker selection (unchanged) -function findBestWorkerVMwareDRS(workers, currentLoads) { - return workers - .map(worker => ({ - worker, - score: (currentLoads.get(worker.url) * 0.4) + // 40% load balance - (worker.cpuUsage * 0.35) + // 35% CPU usage - (worker.memoryUsage * 0.25) // 25% memory usage - })) - .sort((a, b) => a.score - b.score)[0].worker; // Lower score = better -} -``` - -### Simplified Reconciliation Flow -1. **Database Change**: Any process modifies database (Display, Camera, Playlist, Model) -2. **Trigger Notification**: Process sends `regenerate_subscriptions` to `worker:master:commands` -3. **Complete Regeneration**: Master queries database for all active Display+Camera+Playlist combinations -4. **Desired State Creation**: Master generates fresh desired subscriptions from database query results -5. **Diff Analysis**: Master compares fresh desired state vs current actual state on workers -6. **Global Reconciliation**: Master applies VMware DRS algorithm to reconcile differences -7. **Worker Commands**: Master sends subscription/unsubscription commands to workers -8. **State Update**: Master updates actual subscription state in Redis - -### Key Simplifications vs Previous Architecture -- **No Incremental State Management**: No complex tracking of individual subscription changes -- **No State Synchronization Issues**: Desired state always freshly derived from database -- **No Complex Command Protocols**: Only one command type: `regenerate_subscriptions` -- **No Partial Update Bugs**: Complete regeneration eliminates edge cases and race conditions -- **Zero Database-Redis Divergence**: Database is always the authoritative source -- **Simpler Service Layer**: Services just update database + trigger, no subscription logic - -## Class Responsibilities Overview - -### Core Class Functions - -| Class | Primary Responsibility | Key Functions | Process Type | -|-------|----------------------|---------------|--------------| -| **DetectorCluster** | Public API facade and event management | • `subscribeToCamera()` - Legacy interface (triggers regeneration)
• `addDetectionListener()` - Callback registration
• `getState()` - Cluster monitoring
• Event emission to external services | Both Master & Slave | -| **MasterSlaveWorkerCluster** | Pure declarative cluster coordination | **Master**: `regenerateDesiredStateFromDatabase()`, `rebalanceCameraSubscriptions()`, `connectToAllWorkers()`
**Slave**: Minimal role - just routes detection messages
**Both**: `handleDetectionMessage()` for callbacks | Both (different roles) | -| **DetectorConnection** | Individual worker WebSocket management | • `initialize()` - WebSocket connection setup
• `subscribeToCamera()` - Send subscription to worker
• `handleImageDetectionResponse()` - Process AI results
• `resubscribeAll()` - Restore subscriptions after reconnect | Master Only | -| **CameraService** | Database operations + trigger notifications | • `addCamera()` - Database create + trigger regeneration
• `updateCamera()` - Database update + trigger regeneration
• `removeCamera()` - Database delete + trigger regeneration | Both Master & Slave | -| **DisplayService** | Database operations + trigger notifications | • `registerDisplay()` - Database create + trigger regeneration
• `updateDisplay()` - Database update + trigger regeneration
• `deleteDisplay()` - Database delete + trigger regeneration | Both Master & Slave | -| **SubscriptionTrigger** | Simple notification system | • `triggerSubscriptionUpdate()` - Send regeneration request to master | Both Master & Slave | - -## Object Relationship Diagrams - -### Core Class Structure and Methods - -```mermaid -classDiagram - class CameraService { - +addCamera(identifier, rtspUrl) - +removeCamera(identifier) - +resubscribeCamera(identifier) - +getCameras() - +updateCamera(...) - -processDetection(data) - } - - class DetectorCluster { - +initialize() - +subscribeToCamera(...) - +unsubscribeFromCamera(subscriptionId) - +unsubscribeFromAllWithCameraID(cameraId) - +getState() - +addDetectionListener(subscriptionId, callback) - +addGlobalDetectionListener(callback) - -handleWorkerDetection(data) - } - - class MasterSlaveWorkerCluster { - +initialize() - +subscribeToCamera(...) - +storeCameraSubscription(subscription) - +getClusterState() - +shutdown() - -connectToAllWorkers() [MASTER] - -rebalanceCameraSubscriptions() [MASTER] - -triggerRebalancing() [MASTER] - -becomeMaster() - -becomeSlave() - -setupMasterElectionListeners() - } - - class DetectorConnection { - +initialize() - +subscribeToCamera(...) - +unsubscribeFromCamera(subscriptionId) - +getCameraImage(cameraId) - +setSessionId(displayId, sessionId) - +getState() - -connect() - -resubscribeAll() - -handleImageDetectionResponse(data) - -scheduleReconnect() - } - - CameraService --> DetectorCluster : "subscribeToCamera()\ngetState()" - DetectorCluster --> MasterSlaveWorkerCluster : "initialize()\nstoreCameraSubscription()" - MasterSlaveWorkerCluster --> DetectorConnection : "[MASTER] creates connections" -``` - -### Direct Function Call Relationships - -```mermaid -graph TD - API[API Routes] --> CS[CameraService] - CS --> |subscribeToCamera
getState
unsubscribeFromAllWithCameraID| DC[DetectorCluster] - DC --> |initialize
storeCameraSubscription
getClusterState
subscribeToCamera| MSC[MasterSlaveWorkerCluster] - - subgraph "Master Process Only" - MSC --> |connectToAllWorkers
creates connections| CONN[DetectorConnection] - CONN --> |WebSocket calls| PW[Python ML Worker] - end - - ME[MasterElection] --> |getIsMaster
getNodeId
getSlaves| MSC - WL[WorkerLogger] --> |attachToDetectorCluster| DC - - classDef masterOnly fill:#ffcccc - classDef external fill:#ffffcc - - class CONN masterOnly - class PW external - class API external -``` - -### Event-Driven Communication - -```mermaid -graph LR - subgraph "Internal Events" - MSC[MasterSlaveWorkerCluster] -.-> |emit detection| DC[DetectorCluster] - MSC -.-> |emit worker:online
emit worker:offline| DC - DC -.-> |emit worker:detection_result
emit worker:online
emit worker:offline| CS[CameraService] - DC -.-> |emit events| WL[WorkerLogger] - ME[MasterElection] -.-> |master-acquired
master-lost
slave-registered
slave-removed| MSC - end - - subgraph "Callback System" - CS -.-> |callback registration| DC - DC -.-> |detection callbacks| CS - end - - subgraph "WebSocket Events (Master Only)" - CONN[DetectorConnection] -.-> |handleWorkerMessage
handleWorkerOnline
handleWorkerOffline| MSC - PW[Python ML Worker] -.-> |IMAGE_DETECTION
STATE_REPORT| CONN - end - - classDef events fill:#e6f3ff - classDef callbacks fill:#fff2e6 - classDef websocket fill:#ffe6e6 - - class MSC,DC,CS,WL events - class CONN,PW websocket -``` - -### Redis Communication Patterns - -```mermaid -graph TB - subgraph "Master Process" - M[Master MasterSlaveWorkerCluster] - end - - subgraph "Slave Processes" - S1[Slave Process 1] - S2[Slave Process 2] - end - - subgraph "Redis Channels" - SC1[worker:slave:slave1] - SC2[worker:slave:slave2] - MC[worker:master:commands] - AC[worker:assignments:changed] - end - - subgraph "Redis Storage" - WC[worker:connections] - WA[worker:assignments] - WS[worker:slaves] - CS[worker:camera_subscriptions] - end - - M --> |publish detection routing| SC1 - M --> |publish detection routing| SC2 - M --> |publish assignments| AC - M --> |hSet/hGet state| WC - M --> |hSet/hGet assignments| WA - M --> |hSet/hGet subscriptions| CS - - S1 --> |publish commands| MC - S2 --> |publish commands| MC - S1 --> |hSet registration| WS - S2 --> |hSet registration| WS - - SC1 --> |subscribe| S1 - SC2 --> |subscribe| S2 - MC --> |subscribe| M - AC --> |subscribe all| S1 - AC --> |subscribe all| S2 -``` - -## Method Call Flow Analysis - -### Camera Subscription Flow (External Request → Worker) - -```mermaid -sequenceDiagram - participant API as API Routes - participant CS as CameraService - participant DB as Database - participant ST as SubscriptionTrigger - participant R as Redis - participant MSC as MasterSlaveCluster - participant CONN as DetectorConnection - participant W as Python Worker - - Note over API,W: Pure Declarative Flow - API->>+CS: POST /api/camera - CS->>+DB: db.cameraEntity.create({...}) - DB-->>-CS: Camera created - CS->>+ST: triggerSubscriptionUpdate('camera.created', id) - ST->>+R: publish(worker:master:commands, {type: 'regenerate_subscriptions', ...}) - - Note over R,MSC: Only Master Processes Commands - R->>+MSC: Master receives regeneration request - MSC->>+MSC: regenerateDesiredStateFromDatabase() - MSC->>+DB: Query all Display+Camera+Playlist combinations - DB-->>-MSC: Active display configurations - MSC->>+MSC: Generate fresh desired subscriptions - MSC->>+R: Store desired state in Redis - MSC->>+MSC: rebalanceCameraSubscriptions() - MSC->>+MSC: findBestWorkerForSubscription() - MSC->>+CONN: subscribeToCamera(subscriptionId, rtspUrl, ...) - CONN->>+W: WebSocket: {type: "subscribe", payload: {...}} - W-->>-CONN: WebSocket: {type: "stateReport", ...} - CONN->>-MSC: handleWorkerOnline(workerUrl) - MSC->>-R: Update actual subscription state - - Note over W,CS: Detection Processing (unchanged) - W->>CONN: Detection results - CONN->>MSC: Route to assigned slave - MSC->>CS: Detection callback - CS-->>-API: Camera added successfully -``` - -### Detection Processing Flow (Worker → External Callback) - -```mermaid -sequenceDiagram - participant W as Python Worker - participant CONN as DetectorConnection - participant MSC as MasterSlaveCluster - participant R as Redis - participant DC as DetectorCluster - participant CS as CameraService - - Note over W,CS: AI Detection Result Processing - W->>+CONN: WebSocket: {type: "imageDetection", subscriptionIdentifier, data} - CONN->>+MSC: handleWorkerMessage(ImageDetectionResponse) - - Note over MSC: Master finds assigned slave - MSC->>+MSC: findWorkerForSubscription(subscriptionId) - MSC->>+R: hGet(worker:assignments, workerUrl) - MSC->>+R: publish(worker:slave:{slaveId}, {type: 'detection', ...}) - - Note over R: Redis routes to assigned slave - R-->>+MSC: Slave receives detection message - MSC->>+MSC: handleDetectionMessage(message) - MSC->>+DC: emit('detection', detectionData) - - Note over DC: Process detection and trigger callbacks - DC->>+DC: handleWorkerDetection(data) - DC->>+DC: detectionListeners.get(subscriptionId).forEach(callback) - DC->>+CS: callback(detectionData) - DC->>+DC: emit('worker:detection_result', {url, cameraId, detections}) - - Note over CS: External service processes detection - CS->>+CS: processDetection(data) - CS-->>CS: updateAnalytics(), triggerDecisionTrees() -``` - -### Master Election and Failover Flow - -```mermaid -sequenceDiagram - participant ME as MasterElection - participant MSC1 as MasterSlaveCluster (Process 1) - participant MSC2 as MasterSlaveCluster (Process 2) - participant R as Redis - participant W1 as Python Worker 1 - participant W2 as Python Worker 2 - - Note over ME,W2: Master Failover Scenario - - %% Initial state - ME->>+MSC1: emit('master-acquired') - MSC1->>+MSC1: becomeMaster() - ME->>+MSC2: emit('master-lost') - MSC2->>+MSC2: becomeSlave() - - ME->>+R: Automatic slave registration - MSC1->>+W1: WebSocket connection (Master) - MSC1->>+W2: WebSocket connection (Master) - - Note over MSC1: Original master fails - MSC1--xMSC1: Process crash/network failure - - %% MasterElection detects failure and triggers new election - ME->>+ME: Detect failed master, trigger election - ME->>+MSC2: emit('master-acquired') - MSC2->>+MSC2: becomeMaster() - - %% Master recovery process - MSC2->>+MSC2: connectToAllWorkers() - MSC2->>+W1: WebSocket reconnection - MSC2->>+W2: WebSocket reconnection - - MSC2->>+MSC2: healClusterAssignments() - MSC2->>+R: hGetAll(worker:camera_subscriptions) - MSC2->>+MSC2: rebalanceCameraSubscriptions() - - %% Restore subscriptions - MSC2->>+W1: WebSocket: SUBSCRIBE commands - MSC2->>+W2: WebSocket: SUBSCRIBE commands - - Note over MSC2,W2: New master operational - slave registration handled by MasterElection -``` - -## System Architecture Diagrams - -### Master-Slave Cluster Architecture - -```mermaid -graph TB - subgraph "Backend Process Cluster" - M[Master Process
NodeJS Backend] - S1[Slave Process 1
NodeJS Backend] - S2[Slave Process 2
NodeJS Backend] - S3[Slave Process N
NodeJS Backend] - end - - subgraph "Python Worker Cluster" - W1[Python ML Worker 1
WebSocket Server] - W2[Python ML Worker 2
WebSocket Server] - W3[Python ML Worker N
WebSocket Server] - end - - subgraph "Redis Coordination Layer" - R[(Redis)] - R --- C1[worker:slave:* channels] - R --- C2[worker:connections state] - R --- C3[worker:assignments mapping] - R --- C4[worker:camera_subscriptions] - end - - M ===|WebSocket Connections
Only Master| W1 - M ===|WebSocket Connections
Only Master| W2 - M ===|WebSocket Connections
Only Master| W3 - - M <-->|Pub/Sub Messages| R - S1 <-->|Pub/Sub Messages| R - S2 <-->|Pub/Sub Messages| R - S3 <-->|Pub/Sub Messages| R - - M -.->|Route Messages| S1 - M -.->|Route Messages| S2 - M -.->|Route Messages| S3 -``` - -### Data Flow Architecture - -```mermaid -sequenceDiagram - participant CS as CameraService - participant DC as DetectorCluster - participant MS as MasterSlaveCluster - participant R as Redis - participant W as Python Worker - participant S as Slave Process - - Note over CS,S: Camera Subscription Flow - - CS->>DC: subscribeToCamera(cameraId, rtspUrl, modelUrl, ...) - DC->>MS: storeCameraSubscription({...}) - - alt Master Process - MS->>MS: findBestWorkerForSubscription() - MS->>R: hSet(camera_subscriptions, subscriptionId, {...}) - MS->>W: WebSocket: SUBSCRIBE command - W->>MS: STATE_REPORT (subscription confirmed) - MS->>R: publish(worker:slave:{slaveId}, detection_message) - else Slave Process - MS->>R: publish(worker:master:commands, subscribe_command) - Note over MS: Routes to master for execution - end - - Note over CS,S: Detection Processing Flow - - W->>MS: WebSocket: IMAGE_DETECTION response - MS->>MS: findSlaveForWorker(workerUrl) - MS->>R: publish(worker:slave:{slaveId}, detection_data) - R->>S: Redis pub/sub delivery - S->>DC: emit('detection', detectionData) - DC->>CS: callback(detectionData) -``` - -### Subscription Lifecycle Management - -```mermaid -stateDiagram-v2 - [*] --> Pending: Camera Subscription Request - - Pending --> Active: Worker accepts subscription - Pending --> Failed: Worker rejects/unavailable - Pending --> Recovering: Assignment change needed - - Active --> Recovering: Worker goes offline - Active --> [*]: Unsubscribe request - - Recovering --> Active: Reassigned to online worker - Recovering --> Failed: No workers available - Recovering --> [*]: Subscription expired - - Failed --> Recovering: Worker becomes available - Failed --> [*]: Max retries exceeded - - note right of Recovering - Automatic rebalancing every 30s - Master detects offline workers - Reassigns to healthy workers - end note -``` - -### Worker Connection State Machine - -```mermaid -stateDiagram-v2 - [*] --> Connecting: initialize() - - Connecting --> Online: WebSocket connected + STATE_REPORT received - Connecting --> Reconnecting: Connection failed - - Online --> Offline: Heartbeat timeout (10s) - Online --> Reconnecting: WebSocket error/close - Online --> [*]: close() called - - Offline --> Reconnecting: Scheduled reconnect (10s) - Offline --> [*]: close() called - - Reconnecting --> Online: Reconnection successful - Reconnecting --> Reconnecting: Reconnection failed (retry) - Reconnecting --> [*]: close() called - - note right of Online - - Sends heartbeat every 2s - - Processes subscriptions - - Reports resource usage - - Handles detection results - end note -``` - -### Redis Channel Communication Flow - -```mermaid -graph LR - subgraph "Master Process" - M[Master] - WS1[WebSocket to Worker 1] - WS2[WebSocket to Worker 2] - end - - subgraph "Slave Processes" - S1[Slave 1] - S2[Slave 2] - end - - subgraph "Redis Channels" - CH1[worker:slave:slave1] - CH2[worker:slave:slave2] - CH3[worker:messages:upstream] - CH4[worker:assignments:changed] - end - - WS1 -->|Detection Data| M - WS2 -->|Detection Data| M - - M -->|Route by Assignment| CH1 - M -->|Route by Assignment| CH2 - - CH1 -->|Subscribed| S1 - CH2 -->|Subscribed| S2 - - S1 -->|Commands/Responses| CH3 - S2 -->|Commands/Responses| CH3 - CH3 -->|Subscribed| M - - M -->|Assignment Updates| CH4 - CH4 -->|Subscribed| S1 - CH4 -->|Subscribed| S2 -``` - -### Detailed Message Flow by Channel - -```mermaid -graph TB - subgraph "Python ML Workers" - W1[Worker 1
ws://worker1:8000] - W2[Worker 2
ws://worker2:8000] - W3[Worker N
ws://workerN:8000] - end - - subgraph "Master Process (Only One)" - M[Master Backend Process] - subgraph "Master Managed Data" - WC1[WebSocket Connection Pool] - AS[Assignment State] - SUB[Subscription Manager] - end - end - - subgraph "Redis Channels & Storage" - subgraph "Individual Slave Channels" - SC1["worker:slave:slave-uuid-1"] - SC2["worker:slave:slave-uuid-2"] - SC3["worker:slave:slave-uuid-N"] - end - - subgraph "Master Coordination Channels" - MC["worker:master:commands"] - ACH["worker:assignments:changed"] - UPC["worker:messages:upstream"] - SEC["worker:subscription:events"] - end - - subgraph "Persistent Storage" - WCS["worker:connections
(Worker Health States)"] - WAS["worker:assignments
(Worker→Slave Mapping)"] - WSS["worker:slaves
(Slave Registration)"] - CSS["worker:camera_subscriptions
(Subscription Persistence)"] - end - end - - subgraph "Slave Processes" - S1[Slave Process 1
slave-uuid-1] - S2[Slave Process 2
slave-uuid-2] - S3[Slave Process N
slave-uuid-N] - end - - %% WebSocket Communications (Master Only) - W1 -.->|"WebSocket Messages:
• IMAGE_DETECTION
• STATE_REPORT
• PATCH_SESSION"| WC1 - W2 -.->|"WebSocket Messages:
• IMAGE_DETECTION
• STATE_REPORT
• PATCH_SESSION"| WC1 - W3 -.->|"WebSocket Messages:
• IMAGE_DETECTION
• STATE_REPORT
• PATCH_SESSION"| WC1 - - WC1 -.->|"WebSocket Commands:
• SUBSCRIBE
• UNSUBSCRIBE
• REQUEST_STATE
• SET_SESSION_ID"| W1 - WC1 -.->|"WebSocket Commands:
• SUBSCRIBE
• UNSUBSCRIBE
• REQUEST_STATE
• SET_SESSION_ID"| W2 - WC1 -.->|"WebSocket Commands:
• SUBSCRIBE
• UNSUBSCRIBE
• REQUEST_STATE
• SET_SESSION_ID"| W3 - - %% Master Redis Operations - M -->|"hSet() operations:
• Worker states
• Assignments
• Subscriptions"| WCS - M -->|"hSet() operations:
• Worker→Slave mapping
• Load balancing data"| WAS - M -->|"hSet() operations:
• Subscription details
• Assignment tracking"| CSS - - %% Master to Slave Routing - M -->|"Detection Routing:
{type: 'detection',
workerUrl: string,
data: ImageDetectionResponse,
timestamp: string}"| SC1 - M -->|"Detection Routing:
{type: 'detection',
workerUrl: string,
data: ImageDetectionResponse,
timestamp: string}"| SC2 - M -->|"Detection Routing:
{type: 'detection',
workerUrl: string,
data: ImageDetectionResponse,
timestamp: string}"| SC3 - - M -->|"Assignment Updates:
{type: 'assignments_updated',
assignments: Record,
timestamp: string}"| ACH - - %% Slave to Master Communication - S1 -->|"Slave Commands:
{type: 'subscribe_camera',
subscriptionIdentifier: string,
rtspUrl: string,
modelUrl: string,
modelId: number,
snapshotUrl?: string,
cropX1?: number, ...}"| MC - S2 -->|"Slave Commands:
{type: 'subscribe_camera',
subscriptionIdentifier: string,
rtspUrl: string,
modelUrl: string,
modelId: number,
snapshotUrl?: string,
cropX1?: number, ...}"| MC - S3 -->|"Slave Commands:
{type: 'subscribe_camera',
subscriptionIdentifier: string,
rtspUrl: string,
modelUrl: string,
modelId: number,
snapshotUrl?: string,
cropX1?: number, ...}"| MC - - %% Slave Registration and Heartbeats - S1 -->|"hSet() Slave Registration:
{slaveId: string,
processId: string,
online: boolean,
workload: number,
lastSeen: string,
capabilities: {...}}"| WSS - S2 -->|"hSet() Slave Registration:
{slaveId: string,
processId: string,
online: boolean,
workload: number,
lastSeen: string,
capabilities: {...}}"| WSS - S3 -->|"hSet() Slave Registration:
{slaveId: string,
processId: string,
online: boolean,
workload: number,
lastSeen: string,
capabilities: {...}}"| WSS - - %% Channel Subscriptions - SC1 -->|"Subscribed"| S1 - SC2 -->|"Subscribed"| S2 - SC3 -->|"Subscribed"| S3 - - MC -->|"Subscribed"| M - ACH -->|"Subscribed (All Slaves)"| S1 - ACH -->|"Subscribed (All Slaves)"| S2 - ACH -->|"Subscribed (All Slaves)"| S3 - - style M fill:#ff9999 - style WC1 fill:#ffcc99 - style AS fill:#ffcc99 - style SUB fill:#ffcc99 - style S1 fill:#99ccff - style S2 fill:#99ccff - style S3 fill:#99ccff -``` - -### Channel Message Specification - -| Channel Name | Direction | Message Type | Sender | Receiver | Payload Structure | Purpose | -|--------------|-----------|--------------|---------|-----------|-------------------|---------| -| `worker:slave:{slaveId}` | Master→Slave | `detection` | Master Process | Assigned Slave | `{type: 'detection', workerUrl: string, data: ImageDetectionResponse, timestamp: string}` | Route AI detection results from workers to processing slaves | -| `worker:master:commands` | Slave→Master | `regenerate_subscriptions` | Any Process | Master Process | `{type: 'regenerate_subscriptions', reason: string, triggeredBy: string, timestamp: string}` | Notify master that database changed and subscriptions need regeneration | -| `worker:assignments:changed` | Master→All Slaves | `assignments_updated` | Master Process | All Slave Processes | `{type: 'assignments_updated', assignments: Record, timestamp: string}` | Broadcast worker-to-slave assignment changes for rebalancing | -| `worker:messages:upstream` | Slave→Master | Various | Any Slave Process | Master Process | `{type: string, slaveId: string, data: any, timestamp: string}` | General slave-to-master communication (currently unused) | - -### Redis Hash Storage Specification - -| Redis Key | Data Type | Content | Update Pattern | Cleanup Strategy | -|-----------|-----------|---------|----------------|-------------------| -| `worker:connections` | Hash Map | `{[workerUrl]: JSON.stringify(WorkerConnectionState)}` | Master updates every 2s | Manual cleanup only | -| `worker:assignments` | Hash Map | `{[workerUrl]: slaveId}` | Master updates on rebalancing | Manual cleanup only | -| `worker:camera_subscriptions` | Hash Map | `{[subscriptionId]: JSON.stringify(CameraSubscription)}` | Master on subscription changes | Manual cleanup only | -| `master-election:slaves` | Hash Map | `{[nodeId]: JSON.stringify(SlaveNode)}` | MasterElection service manages | TTL-based cleanup | - -### WebSocket Message Protocol - -| Direction | Message Type | JSON Structure | Trigger | Response Expected | -|-----------|--------------|----------------|---------|-------------------| -| Backend→Worker | `SUBSCRIBE` | `{type: "subscribe", payload: {subscriptionIdentifier, rtspUrl, snapshotUrl?, snapshotInterval?, modelUrl, modelName, modelId, cropX1?, cropY1?, cropX2?, cropY2?}}` | Camera subscription request | STATE_REPORT confirmation | -| Backend→Worker | `UNSUBSCRIBE` | `{type: "unsubscribe", payload: {subscriptionIdentifier}}` | Camera unsubscription | STATE_REPORT confirmation | -| Backend→Worker | `REQUEST_STATE` | `{type: "requestState"}` | Health check or monitoring | STATE_REPORT response | -| Backend→Worker | `SET_SESSION_ID` | `{type: "setSessionId", payload: {displayIdentifier, sessionId}}` | Associate session with display | None | -| Backend→Worker | `PATCH_SESSION_RESULT` | `{type: "patchSessionResult", payload: {sessionId, success, message?}}` | Session update response | None | -| Worker→Backend | `IMAGE_DETECTION` | `{type: "imageDetection", subscriptionIdentifier, timestamp, data: {detection: {carModel?, carBrand?, carYear?, bodyType?, licensePlateText?, licensePlateType?}, modelId, modelName}}` | AI detection result | None | -| Worker→Backend | `STATE_REPORT` | `{type: "stateReport", cpuUsage, memoryUsage, gpuUsage?, gpuMemoryUsage?, cameraConnections: [{subscriptionIdentifier, modelId, modelName, online, cropX?, cropY?}]}` | Periodic health report (every 2s) | None | -| Worker→Backend | `PATCH_SESSION` | `{type: "patchSession", sessionId, data: any}` | Session data update from ML processing | PATCH_SESSION_RESULT | - -## Event System Architecture - -### Event Flow Hierarchy - -```mermaid -graph TD - subgraph "Service Layer" - CS[CameraService] - end - - subgraph "Cluster Layer" - DC[DetectorCluster] - DC --> DCE[Detection Events] - DC --> WOE[Worker Online Events] - DC --> WOFE[Worker Offline Events] - end - - subgraph "Worker Management Layer" - MS[MasterSlaveWorkerCluster] - MS --> DE[detection] - MS --> WC[worker:connected] - MS --> WD[worker:disconnected] - MS --> WSE[worker:websocket_error] - MS --> WON[worker:online] - MS --> WOFF[worker:offline] - MS --> WSR[worker:state_report] - end - - subgraph "Connection Layer" - DConn[DetectorConnection] - DConn --> IMG[IMAGE_DETECTION] - DConn --> STATE[STATE_REPORT] - DConn --> PATCH[PATCH_SESSION] - end - - DConn --> MS - MS --> DC - DC --> CS - - IMG -.-> DE - STATE -.-> WSR - WC -.-> WOE - WD -.-> WOFE -``` - -### Message Types and Routing - -#### WebSocket Message Types (Python Worker → Backend) -- `IMAGE_DETECTION`: AI detection results from camera streams -- `STATE_REPORT`: Worker health, resource usage, and subscription status -- `PATCH_SESSION`: Session data updates from worker processing - -#### Redis Channel Message Types -- `detection`: Detection results routed from master to assigned slave -- `command_response`: Command acknowledgment and status updates -- `heartbeat`: Worker and slave health monitoring messages -- `assignments_updated`: Worker-to-slave assignment change notifications - -#### Internal Event Types -- `worker:online`: Worker connection established and ready -- `worker:offline`: Worker connection lost or health check failed -- `worker:connected`: WebSocket connection opened (not necessarily ready) -- `worker:disconnected`: WebSocket connection closed -- `worker:websocket_error`: WebSocket communication errors -- `worker:detection_result`: Processed detection with metadata -- `worker:state_report`: Worker resource and subscription status - -## Subscription Management - -### Camera Subscription Flow - -1. **Registration Phase** - - `CameraService.subscribeToCamera()` → `DetectorCluster.subscribeToCamera()` - - Master process finds optimal worker using load balancing algorithm - - Subscription stored in Redis with full configuration including crop parameters - - Master sends WebSocket SUBSCRIBE command to assigned worker - -2. **Processing Phase** - - Python worker establishes RTSP connection to camera - - Worker performs AI inference on video stream frames - - Detection results sent back via WebSocket with subscription identifier - - Master routes results to appropriate slave based on worker assignments - -3. **Rebalancing Phase** - - Master monitors worker health every 30 seconds - - Orphaned subscriptions (offline workers) automatically detected - - Load balancing algorithm reassigns cameras to healthy workers - - Fresh model URLs generated to handle S3 presigned URL expiration - -### Load Balancing Algorithm - -```typescript -// Simplified load balancing logic -function findBestWorkerForSubscription(onlineWorkers, allSubscriptions) { - return onlineWorkers - .sort((a, b) => { - const loadA = getSubscriptionCount(a.url); - const loadB = getSubscriptionCount(b.url); - if (loadA !== loadB) { - return loadA - loadB; // Prefer lower load - } - return (a.cpuUsage || 0) - (b.cpuUsage || 0); // Then prefer lower CPU - })[0]; -} -``` - -### Automatic Failover Process - -1. **Detection**: Master detects worker offline via missed heartbeats (10s timeout) -2. **Identification**: System identifies all camera subscriptions assigned to offline worker -3. **Reassignment**: Load balancer selects optimal replacement worker -4. **Migration**: Subscription updated in Redis with new worker assignment -5. **Resubscription**: Master sends SUBSCRIBE command to new worker with fresh model URL -6. **Verification**: New worker confirms subscription and begins processing - -## Resource Management - -### Connection Pooling -- Master maintains persistent WebSocket connections to all configured workers -- Connection sharing across all backend processes reduces resource overhead -- Automatic reconnection with exponential backoff prevents connection storms - -### Memory Management -- Redis data uses manual cleanup to prevent accidental state loss -- Subscription callbacks stored in local memory with automatic cleanup on unsubscribe -- Worker resource usage tracked in real-time to prevent overload - -### CPU and GPU Monitoring -- Workers report resource usage every 2 seconds via STATE_REPORT messages -- Load balancing algorithm considers CPU usage when assigning new subscriptions -- GPU utilization tracked for ML model optimization and capacity planning - -## Error Handling - -### Connection Error Recovery -- **Exponential Backoff**: 10-second fixed interval reconnection attempts -- **Circuit Breaker**: Automatic failover prevents overwhelming failed workers -- **Graceful Degradation**: System continues operating with available workers - -### Master Election Failover -- **Leadership Transfer**: New master elected via Redis-based coordination -- **State Recovery**: Worker connections and subscriptions restored from Redis persistence -- **Seamless Transition**: No subscription loss during master failover process - -### Monitoring and Observability - -#### Structured Logging Topics -- `detector-cluster`: High-level cluster operations and state changes -- `master-slave-worker-cluster`: Worker assignment and rebalancing operations -- `DetectorConnection`: WebSocket connection events and message processing - -#### Monitoring Information -- Subscription identifier format: `${displayId};${cameraId}` for traceability -- Worker assignment tracking with process ID and timestamp correlation -- Redis pub/sub message routing with structured logging -- Heartbeat and health check timing with millisecond precision - -## Configuration Parameters - -### Timing Configuration -```typescript -const WORKER_TIMEOUT_MS = 10000; // Worker heartbeat timeout -const SLAVE_HEARTBEAT_INTERVAL = 5000; // Slave heartbeat frequency -const SLAVE_TIMEOUT = 15000; // Slave registration timeout -const REBALANCE_INTERVAL = 30000; // Automatic rebalancing frequency -const STATE_UPDATE_INTERVAL = 2000; // Worker state update frequency -const RECONNECT_DELAY = 10000; // WebSocket reconnection delay -``` - -### Environment Variables -```bash -DETECTOR_WORKERS=ws://worker1:8000,ws://worker2:8000 # Python worker URLs -REDIS_HOST=localhost # Redis coordination server -REDIS_PORT=6379 # Redis server port -REDIS_PASSWORD=secure_password # Redis authentication -DETECT_DEBUG=true # Enable detailed structured logging -``` - -## Performance Characteristics - -### Scalability Metrics -- **Horizontal Scaling**: Add backend processes without WebSocket connection changes -- **Worker Scaling**: Python ML workers scale independently of backend processes -- **Redis Optimization**: Efficient pub/sub routing with minimal memory overhead - -### Throughput Capabilities -- **Camera Subscriptions**: Support for 100+ simultaneous camera streams per worker -- **Detection Processing**: Sub-second AI inference with real-time result delivery -- **Message Routing**: Sub-millisecond Redis pub/sub message delivery - -### Resource Efficiency -- **Connection Multiplexing**: Single WebSocket per worker shared across all processes -- **Memory Usage**: Lightweight subscription state with callback cleanup -- **Network Optimization**: Binary WebSocket frames with JSON payload compression - -## Public Interface Specification - -The distributed worker cluster exposes a clean, simplified interface to external services like CameraService, hiding the complexity of the underlying master-slave architecture. All interactions go through the `DetectorCluster` class, which serves as the primary facade. - -### Primary Interface: DetectorCluster - -The `DetectorCluster` class in `/services/DetectorCluster.ts` provides the main public interface that external services interact with. It abstracts away the distributed architecture complexity and provides consistent behavior regardless of whether the current process is a master or slave. - -#### Core Interface Methods - -##### Camera Subscription Management - -```typescript -/** - * Subscribe to a camera stream for AI detection processing - * @param subscriptionIdentifier - Unique identifier format: "${displayId};${cameraId}" - * @param rtspUrl - RTSP stream URL for the camera - * @param modelUrl - Pre-signed S3 URL for AI model (1hr TTL) - * @param modelId - Database ID of the AI model - * @param modelName - Human-readable model identifier - * @param callback - Function called when detection results are received - * @param snapshotUrl - Optional HTTP endpoint for camera snapshots - * @param snapshotInterval - Optional snapshot capture interval in milliseconds - * @param cropX1, cropY1, cropX2, cropY2 - Optional image crop coordinates - * @returns Promise - Always returns true (errors thrown as exceptions) - */ -public async subscribeToCamera( - subscriptionIdentifier: string, - rtspUrl: string, - modelUrl: string, - modelId: number, - modelName: string, - callback: Function, - snapshotUrl?: string, - snapshotInterval?: number, - cropX1?: number, - cropY1?: number, - cropX2?: number, - cropY2?: number -): Promise -``` - -**Behavior:** -- **Master Process**: Stores subscription in Redis, assigns to optimal worker, sends WebSocket command -- **Slave Process**: Routes subscription request to master via Redis pub/sub -- **Callback Registration**: Stores callback locally for detection result processing -- **Persistence**: All subscription details stored in Redis for failover recovery -- **Load Balancing**: Automatically selects best available worker based on CPU and subscription load - -```typescript -/** - * Unsubscribe from a specific camera stream - * @param subscriptionIdentifier - The subscription to remove - * @returns Promise - Success status - */ -public async unsubscribeFromCamera(subscriptionIdentifier: string): Promise -``` - -**Behavior:** -- Removes local callback listeners immediately -- Subscription cleanup handled automatically by cluster rebalancing -- Safe to call multiple times (idempotent operation) - -```typescript -/** - * Remove all subscriptions for a specific camera across all displays - * @param cameraIdentifier - The camera ID to unsubscribe from all displays - * @returns Promise - */ -public async unsubscribeFromAllWithCameraID(cameraIdentifier: string): Promise -``` - -**Behavior:** -- Finds all subscription identifiers matching pattern `*;${cameraIdentifier}` -- Removes all local callbacks for matched subscriptions -- Cluster automatically handles worker-side cleanup - -##### Event Registration and Callbacks - -```typescript -/** - * Register a callback for detection results from a specific subscription - * @param subscriptionIdentifier - Target subscription - * @param callback - Function to call with detection data - */ -public addDetectionListener(subscriptionIdentifier: string, callback: Function): void - -/** - * Register a global callback for all detection results - * @param callback - Function to call with any detection data - */ -public addGlobalDetectionListener(callback: Function): void -``` - -**Detection Callback Signature:** -```typescript -type DetectionCallback = (data: { - subscriptionIdentifier: string; - timestamp: Date; - data: { - detection: { - carModel?: string; - carBrand?: string; - carYear?: number; - bodyType?: string; - licensePlateText?: string; - licensePlateType?: string; - }; - modelId: number; - modelName: string; - }; -}) => void; -``` - -##### Cluster State Management - -```typescript -/** - * Get comprehensive cluster state for monitoring and status reporting - * @returns Promise - */ -public async getState(): Promise - -/** - * Legacy method - rebalancing now happens automatically - * @returns Promise - Always returns true - */ -public async rebalanceWorkers(): Promise -``` - -**DetectorClusterState Interface:** -```typescript -interface DetectorClusterState { - processId: string; // Current process identifier - isMaster: boolean; // Whether this process is the master - slaveId: string; // This process's slave identifier - totalWorkers: number; // Number of Python ML workers - totalSlaves: number; // Number of backend slave processes - workers: WorkerState[]; // Detailed worker health and status - slaves: SlaveInfo[]; // Slave process information - assignments: Record; // workerUrl -> slaveId mapping -} -``` - -##### Session Management (Future Implementation) - -```typescript -/** - * Associate a session ID with a camera subscription for tracking - * @param subscriptionIdentifier - Target subscription - * @param sessionId - Session ID to associate (null to clear) - * @returns Promise - Success status - */ -public async setSessionId(subscriptionIdentifier: string, sessionId: number | null): Promise - -/** - * Get current camera image via worker REST API - * @param cameraIdentifier - Camera to capture from - * @returns Promise - JPEG image data - */ -public async getCameraImage(cameraIdentifier: string): Promise -``` - -**Note:** These methods are currently not fully implemented in master-slave mode. - -### Event System Interface - -The cluster emits events that external services can listen to for system monitoring and integration: - -#### Emitted Events - -```typescript -// Detection result processed -detectorCluster.on('worker:detection_result', (event: { - url: string; // Worker URL (always 'cluster-managed') - cameraId: string; // Subscription identifier - detections: number; // Number of objects detected (0 or 1) -}) => void); - -// Worker status changes -detectorCluster.on('worker:online', (event: { url: string }) => void); -detectorCluster.on('worker:offline', (event: { url: string }) => void); - -// Connection events -detectorCluster.on('worker:connecting', (event: { url: string }) => void); -detectorCluster.on('worker:disconnected', (event: { url: string, reason: string }) => void); -detectorCluster.on('worker:websocket_error', (event: { url: string, error: string }) => void); -``` - -### Usage Examples - -#### Basic Camera Subscription (CameraService Integration) - -```typescript -import { detectorCluster } from '~/modules/camera/services/CameraService'; - -// Subscribe to camera with AI detection -const success = await detectorCluster.subscribeToCamera( - `display-123;camera-456`, // subscriptionIdentifier - 'rtsp://192.168.1.100:554/stream1', // rtspUrl - 'https://s3.bucket.com/model.onnx', // modelUrl (pre-signed) - 42, // modelId - 'vehicle-detection-v2', // modelName - (detectionData) => { // callback - console.log('Detection:', detectionData.data.detection); - // Process car model, license plate, etc. - }, - 'http://192.168.1.100/snapshot.jpg', // snapshotUrl (optional) - 5000, // snapshotInterval (optional) - 100, 50, 800, 600 // crop coordinates (optional) -); -``` - -#### Event Monitoring Integration - -```typescript -// Monitor worker health -detectorCluster.on('worker:online', (event) => { - console.log(`Worker ${event.url} came online`); - // Update dashboard, send notifications, etc. -}); - -detectorCluster.on('worker:offline', (event) => { - console.log(`Worker ${event.url} went offline`); - // Alert administrators, trigger failover procedures -}); - -// Monitor detection activity -detectorCluster.on('worker:detection_result', (event) => { - if (event.detections > 0) { - console.log(`Camera ${event.cameraId} detected objects`); - // Trigger content changes, log analytics, etc. - } -}); -``` - -#### Cluster State Monitoring - -```typescript -// Get comprehensive cluster status -const state = await detectorCluster.getState(); - -console.log(`Process ${state.processId} is ${state.isMaster ? 'MASTER' : 'SLAVE'}`); -console.log(`Cluster: ${state.totalWorkers} workers, ${state.totalSlaves} slaves`); - -// Monitor worker health -state.workers.forEach(worker => { - console.log(`Worker ${worker.url}: ${worker.online ? 'ONLINE' : 'OFFLINE'}`); - console.log(` CPU: ${worker.cpuUsage}%, Memory: ${worker.memoryUsage}%`); - console.log(` Subscriptions: ${worker.subscriptionCount}`); -}); - -// Check assignments -Object.entries(state.assignments).forEach(([workerUrl, slaveId]) => { - console.log(`Worker ${workerUrl} assigned to slave ${slaveId}`); -}); -``` - -#### Bulk Camera Management - -```typescript -// Remove all subscriptions for a camera being deleted -await detectorCluster.unsubscribeFromAllWithCameraID('camera-456'); - -// Re-subscribe camera to all displays after configuration change -const displays = await getDisplaysForCamera('camera-456'); -for (const display of displays) { - await detectorCluster.subscribeToCamera( - `${display.id};camera-456`, - camera.rtspUrl, - freshModelUrl, - modelId, - modelName, - createDetectionHandler(display.id, camera.id), - camera.snapshotUrl, - camera.snapshotInterval, - display.cropX1, display.cropY1, - display.cropX2, display.cropY2 - ); -} -``` - -### Error Handling Interface - -The cluster interface follows consistent error handling patterns: - -#### Exception Types - -```typescript -// Subscription errors -try { - await detectorCluster.subscribeToCamera(...); -} catch (error) { - // Possible errors: - // - "No workers available for assignment" - // - "Invalid subscription identifier format" - // - "Model URL expired or inaccessible" - // - Redis connection errors -} - -// State retrieval errors -try { - const state = await detectorCluster.getState(); -} catch (error) { - // Returns safe default state on errors - // Logs detailed error information -} -``` - -#### Graceful Degradation - -- **No Workers Available**: Subscriptions stored in Redis, will activate when workers come online -- **Master Process Failure**: New master elected, all subscriptions restored from Redis -- **Redis Connection Issues**: Local callbacks continue working, subscriptions restored when connection recovers -- **Invalid Parameters**: Clear error messages with parameter validation - -### Integration Patterns - -#### Service Layer Integration - -```typescript -// CameraService.ts example -export class CameraService { - constructor() { - // Initialize cluster connection - detectorCluster.initialize(); - - // Set up global detection processing - detectorCluster.addGlobalDetectionListener(this.processDetection.bind(this)); - } - - async subscribeCamera(displayId: string, camera: CameraEntity) { - const subscriptionId = `${displayId};${camera.cameraIdentifier}`; - - return await detectorCluster.subscribeToCamera( - subscriptionId, - camera.rtspUrl, - await this.getModelUrl(camera.modelId), - camera.modelId, - camera.modelName, - (data) => this.handleDetection(displayId, camera.id, data), - camera.snapshotUrl, - camera.snapshotInterval, - camera.cropX1, camera.cropY1, - camera.cropX2, camera.cropY2 - ); - } - - private processDetection(data: ImageDetectionResponse) { - // Global detection processing logic - this.updateAnalytics(data); - this.triggerDecisionTrees(data); - } -} -``` - -### Interface Guarantees and Contracts - -#### Reliability Guarantees - -- **At-Least-Once Detection Delivery**: Detection callbacks will be called at least once per detection -- **Subscription Persistence**: Subscriptions survive process restarts and master failovers -- **Automatic Reconnection**: Workers automatically reconnect with exponential backoff -- **Load Balancing**: New subscriptions automatically assigned to least loaded workers - -#### Performance Characteristics - -- **Subscription Latency**: < 100ms for new camera subscriptions -- **Detection Latency**: < 50ms from worker to callback (excluding AI processing time) -- **State Query Performance**: < 10ms for cluster state retrieval -- **Memory Usage**: O(n) where n = number of active subscriptions - -#### Thread Safety - -- **Callback Execution**: All callbacks executed on main event loop (Node.js single-threaded) -- **Concurrent Subscriptions**: Multiple simultaneous subscriptions handled safely -- **State Consistency**: Redis operations use atomic transactions where needed - -This interface specification provides external services with a clear understanding of how to integrate with the distributed worker cluster while maintaining abstraction from the underlying complexity. - -## Architecture Evolution: From Complex to Pure Declarative - -### Previous Architecture Limitations (Addressed) -- **Complex State Synchronization**: Incremental updates between database, Redis desired state, and worker actual state created synchronization complexity -- **Command Protocol Complexity**: Multiple command types (`subscribe_camera`, `unsubscribe_camera`) with complex payloads and error handling -- **State Divergence**: Database and Redis desired state could diverge, causing inconsistent behavior -- **Partial Update Complexity**: Complex logic for handling individual subscription changes led to edge cases and race conditions -- **Service Layer Complexity**: Camera/Display services contained complex subscription management logic - -### Current Pure Declarative Architecture Benefits -- **Single Source of Truth**: Database is the only source for desired state - no secondary state stores to synchronize -- **Zero State Divergence**: Desired state is always freshly derived from database queries, eliminating synchronization complexity -- **Simplified Protocol**: Only one command type (`regenerate_subscriptions`) with minimal payload -- **Consistent State Management**: Complete regeneration eliminates all edge cases and partial update complexity -- **Service Layer Simplicity**: Services just update database + trigger regeneration - no subscription logic -- **Operational Resilience**: System is self-healing and predictable - any database change triggers complete reconciliation - -### VMware DRS-like Benefits -- **Global Optimization**: Every regeneration considers all subscriptions globally for optimal load balancing -- **Automatic Recovery**: System automatically heals from any inconsistent state by regenerating from database -- **Resource Efficiency**: Workers assigned based on real-time CPU/memory metrics with load balancing -- **Fault Tolerance**: Complete state recovery from database after any failure (process crashes, network interruptions, etc.) - -### Performance Characteristics -- **Regeneration Speed**: Database queries are fast (~10ms) even with hundreds of displays -- **Reconciliation Efficiency**: Only changed subscriptions are actually modified on workers -- **Memory Efficiency**: No persistent state storage outside of database and current worker assignments -- **Network Efficiency**: Minimal command protocol reduces Redis pub/sub overhead - -This pure declarative architecture provides the reliability and simplicity of Kubernetes-style declarative resource management while maintaining the performance and scalability needed for real-time camera processing systems. \ No newline at end of file diff --git a/requirements.base.txt b/requirements.base.txt index af22160..b846ae9 100644 --- a/requirements.base.txt +++ b/requirements.base.txt @@ -1,7 +1,7 @@ -torch -torchvision -ultralytics -opencv-python -scipy -filterpy -psycopg2-binary \ No newline at end of file +torch>=1.12.0,<2.1.0 +torchvision>=0.13.0,<0.16.0 +ultralytics>=8.0.0,<8.1.0 +opencv-python>=4.6.0,<4.9.0 +scipy>=1.9.0,<1.12.0 +filterpy>=1.4.0,<1.5.0 +psycopg2-binary>=2.9.0,<2.10.0 \ No newline at end of file diff --git a/siwatsystem/pympta.py b/siwatsystem/pympta.py index fd1485d..d21232d 100644 --- a/siwatsystem/pympta.py +++ b/siwatsystem/pympta.py @@ -514,65 +514,6 @@ def resolve_field_mapping(value_template, branch_results, action_context): logger.error(f"Error resolving field mapping '{value_template}': {e}") return None -def validate_pipeline_execution(node, regions_dict): - """ - Pre-validate that all required branches will execute successfully before - committing to Redis actions and database records. - - Returns: - - (True, []) if pipeline can execute completely - - (False, missing_branches) if some required branches won't execute - """ - # Get all branches that parallel actions are waiting for - required_branches = set() - - for action in node.get("parallelActions", []): - if action.get("type") == "postgresql_update_combined": - wait_for_branches = action.get("waitForBranches", []) - required_branches.update(wait_for_branches) - - if not required_branches: - # No parallel actions requiring specific branches - logger.debug("No parallel actions with waitForBranches - validation passes") - return True, [] - - logger.debug(f"Pre-validation: checking if required branches {list(required_branches)} will execute") - - # Check each required branch - missing_branches = [] - - for branch in node.get("branches", []): - branch_id = branch["modelId"] - - if branch_id not in required_branches: - continue # This branch is not required by parallel actions - - # Check if this branch would be triggered - trigger_classes = branch.get("triggerClasses", []) - min_conf = branch.get("minConfidence", 0) - - branch_triggered = False - for det_class in regions_dict: - det_confidence = regions_dict[det_class]["confidence"] - - if (det_class in trigger_classes and det_confidence >= min_conf): - branch_triggered = True - logger.debug(f"Pre-validation: branch {branch_id} WILL be triggered by {det_class} (conf={det_confidence:.3f} >= {min_conf})") - break - - if not branch_triggered: - missing_branches.append(branch_id) - logger.warning(f"Pre-validation: branch {branch_id} will NOT be triggered - no matching classes or insufficient confidence") - logger.debug(f" Required: {trigger_classes} with min_conf={min_conf}") - logger.debug(f" Available: {[(cls, regions_dict[cls]['confidence']) for cls in regions_dict]}") - - if missing_branches: - logger.error(f"Pipeline pre-validation FAILED: required branches {missing_branches} will not execute") - return False, missing_branches - else: - logger.info(f"Pipeline pre-validation PASSED: all required branches {list(required_branches)} will execute") - return True, [] - def run_pipeline(frame, node: dict, return_bbox: bool=False, context=None): """ Enhanced pipeline that supports: @@ -705,14 +646,6 @@ def run_pipeline(frame, node: dict, return_bbox: bool=False, context=None): else: logger.debug("No multi-class validation - proceeding with all detections") - # ─── Pre-validate pipeline execution ──────────────────────── - pipeline_valid, missing_branches = validate_pipeline_execution(node, regions_dict) - - if not pipeline_valid: - logger.error(f"Pipeline execution validation FAILED - required branches {missing_branches} cannot execute") - logger.error("Aborting pipeline: no Redis actions or database records will be created") - return (None, None) if return_bbox else None - # ─── Execute actions with region information ──────────────── detection_result = { "detections": all_detections, @@ -853,11 +786,9 @@ def run_pipeline(frame, node: dict, return_bbox: bool=False, context=None): primary_detection = max(all_detections, key=lambda x: x["confidence"]) primary_bbox = primary_detection["bbox"] - # Add branch results and session_id to primary detection for compatibility + # Add branch results to primary detection for compatibility if "branch_results" in detection_result: primary_detection["branch_results"] = detection_result["branch_results"] - if "session_id" in detection_result: - primary_detection["session_id"] = detection_result["session_id"] return (primary_detection, primary_bbox) if return_bbox else primary_detection diff --git a/worker.md b/worker.md index c485db5..302c8ce 100644 --- a/worker.md +++ b/worker.md @@ -2,6 +2,12 @@ This document outlines the WebSocket-based communication protocol between the CMS backend and a detector worker. As a worker developer, your primary responsibility is to implement a WebSocket server that adheres to this protocol. +The current Python Detector Worker implementation supports advanced computer vision pipelines with: +- Multi-class YOLO detection with parallel processing +- PostgreSQL database integration with automatic schema management +- Redis integration for image storage and pub/sub messaging +- Hierarchical pipeline execution with detection → classification branching + ## 1. Connection The worker must run a WebSocket server, preferably on port `8000`. The backend system, which is managed by a container orchestration service, will automatically discover and establish a WebSocket connection to your worker. @@ -25,14 +31,34 @@ To enable modularity and dynamic configuration, the backend will send you a URL 2. Extracting its contents. 3. Interpreting the contents to configure its internal pipeline. -**The contents of the `.mpta` file are entirely up to the user who configures the model in the CMS.** This allows for maximum flexibility. For example, the archive could contain: +**The current implementation supports comprehensive pipeline configurations including:** -- AI/ML Models: Pre-trained models for libraries like TensorFlow, PyTorch, or ONNX. -- Configuration Files: A `config.json` or `pipeline.yaml` that defines a sequence of operations, specifies model paths, or sets detection thresholds. -- Scripts: Custom Python scripts for pre-processing or post-processing. -- API Integration Details: A JSON file with endpoint information and credentials for interacting with third-party detection services. +- **AI/ML Models**: YOLO models (.pt files) for detection and classification +- **Pipeline Configuration**: `pipeline.json` defining hierarchical detection→classification workflows +- **Multi-class Detection**: Simultaneous detection of multiple object classes (e.g., Car + Frontal) +- **Parallel Processing**: Concurrent execution of classification branches with ThreadPoolExecutor +- **Database Integration**: PostgreSQL configuration for automatic table creation and updates +- **Redis Actions**: Image storage with region cropping and pub/sub messaging +- **Dynamic Field Mapping**: Template-based field resolution for database operations -Essentially, the `.mpta` file is a self-contained package that tells your worker _how_ to process the video stream for a given subscription. +**Enhanced MPTA Structure:** +``` +pipeline.mpta/ +├── pipeline.json # Main configuration with redis/postgresql settings +├── car_detection.pt # Primary YOLO detection model +├── brand_classifier.pt # Classification model for car brands +├── bodytype_classifier.pt # Classification model for body types +└── ... +``` + +The `pipeline.json` now supports advanced features like: +- Multi-class detection with `expectedClasses` validation +- Parallel branch processing with `parallel: true` +- Database actions with `postgresql_update_combined` +- Redis actions with region-specific image cropping +- Branch synchronization with `waitForBranches` + +Essentially, the `.mpta` file is a self-contained package that tells your worker *how* to process the video stream for a given subscription, including complex multi-stage AI pipelines with database persistence. ## 4. Messages from Worker to Backend @@ -79,6 +105,15 @@ Sent when the worker detects a relevant object. The `detection` object should be - **Type:** `imageDetection` +**Enhanced Detection Capabilities:** + +The current implementation supports multi-class detection with parallel classification processing. When a vehicle is detected, the system: + +1. **Multi-Class Detection**: Simultaneously detects "Car" and "Frontal" classes +2. **Parallel Processing**: Runs brand and body type classification concurrently +3. **Database Integration**: Automatically creates and updates PostgreSQL records +4. **Redis Storage**: Saves cropped frontal images with expiration + **Payload Example:** ```json @@ -88,19 +123,38 @@ Sent when the worker detects a relevant object. The `detection` object should be "timestamp": "2025-07-14T12:34:56.789Z", "data": { "detection": { - "carModel": "Civic", + "class": "Car", + "confidence": 0.92, "carBrand": "Honda", - "carYear": 2023, + "carModel": "Civic", "bodyType": "Sedan", - "licensePlateText": "ABCD1234", - "licensePlateConfidence": 0.95 + "branch_results": { + "car_brand_cls_v1": { + "class": "Honda", + "confidence": 0.89, + "brand": "Honda" + }, + "car_bodytype_cls_v1": { + "class": "Sedan", + "confidence": 0.85, + "body_type": "Sedan" + } + } }, "modelId": 101, - "modelName": "US-LPR-and-Vehicle-ID" + "modelName": "Car Frontal Detection V1" } } ``` +**Database Integration:** + +Each detection automatically: +- Creates a record in `gas_station_1.car_frontal_info` table +- Generates a unique `session_id` for tracking +- Updates the record with classification results after parallel processing completes +- Stores cropped frontal images in Redis with the session_id as key + ### 4.3. Patch Session > **Note:** Patch messages are only used when the worker can't keep up and needs to retroactively send detections. Normally, detections should be sent in real-time using `imageDetection` messages. Use `patchSession` only to update session data after the fact. @@ -117,9 +171,9 @@ Allows the worker to request a modification to an active session's data. The `da "sessionId": 12345, "data": { "currentCar": { - "carModel": "Civic", - "carBrand": "Honda", - "licensePlateText": "ABCD1234" + "carModel": "Civic", + "carBrand": "Honda", + "licensePlateText": "ABCD1234" } } } @@ -133,33 +187,24 @@ The `data` object in the `patchSession` message is merged with the existing `Dis ```typescript interface DisplayPersistentData { - progressionStage: - | 'welcome' - | 'car_fueling' - | 'car_waitpayment' - | 'car_postpayment' - | null; - qrCode: string | null; - adsPlayback: { - playlistSlotOrder: number; // The 'order' of the current slot - adsId: number | null; - adsUrl: string | null; - } | null; - currentCar: { - carModel?: string; - carBrand?: string; - carYear?: number; - bodyType?: string; - licensePlateText?: string; - licensePlateType?: string; - } | null; - fuelPump: { - /* FuelPumpData structure */ - } | null; - weatherData: { - /* WeatherResponse structure */ - } | null; - sessionId: number | null; + progressionStage: "welcome" | "car_fueling" | "car_waitpayment" | "car_postpayment" | null; + qrCode: string | null; + adsPlayback: { + playlistSlotOrder: number; // The 'order' of the current slot + adsId: number | null; + adsUrl: string | null; + } | null; + currentCar: { + carModel?: string; + carBrand?: string; + carYear?: number; + bodyType?: string; + licensePlateText?: string; + licensePlateType?: string; + } | null; + fuelPump: { /* FuelPumpData structure */ } | null; + weatherData: { /* WeatherResponse structure */ } | null; + sessionId: number | null; } ``` @@ -212,7 +257,7 @@ Instructs the worker to process a camera's RTSP stream using the configuration f > - Capture each snapshot only once per cycle, and reuse it for all display subscriptions sharing that camera. > - Capture each frame/image only once per cycle. > - Reuse the same captured image and snapshot for all display subscriptions that share the camera, processing and routing detection results separately for each display as needed. -> This avoids unnecessary load and bandwidth usage, and ensures consistent detection results and snapshots across all displays sharing the same camera. +> This avoids unnecessary load and bandwidth usage, and ensures consistent detection results and snapshots across all displays sharing the same camera. ### 5.2. Unsubscribe from Camera @@ -324,7 +369,7 @@ This section shows a typical sequence of messages between the backend and the wo > **Note:** Unsubscribe is triggered when a user removes a camera or when the node is too heavily loaded and needs rebalancing. 1. **Connection Established** & **Heartbeat** - - **Worker -> Backend** + * **Worker -> Backend** ```json { "type": "stateReport", @@ -336,7 +381,7 @@ This section shows a typical sequence of messages between the backend and the wo } ``` 2. **Backend Subscribes Camera** - - **Backend -> Worker** + * **Backend -> Worker** ```json { "type": "subscribe", @@ -350,7 +395,7 @@ This section shows a typical sequence of messages between the backend and the wo } ``` 3. **Worker Acknowledges in Heartbeat** - - **Worker -> Backend** + * **Worker -> Backend** ```json { "type": "stateReport", @@ -369,7 +414,7 @@ This section shows a typical sequence of messages between the backend and the wo } ``` 4. **Worker Detects a Car** - - **Worker -> Backend** + * **Worker -> Backend** ```json { "type": "imageDetection", @@ -388,7 +433,7 @@ This section shows a typical sequence of messages between the backend and the wo } } ``` - - **Worker -> Backend** + * **Worker -> Backend** ```json { "type": "imageDetection", @@ -407,7 +452,7 @@ This section shows a typical sequence of messages between the backend and the wo } } ``` - - **Worker -> Backend** + * **Worker -> Backend** ```json { "type": "imageDetection", @@ -427,7 +472,7 @@ This section shows a typical sequence of messages between the backend and the wo } ``` 5. **Backend Unsubscribes Camera** - - **Backend -> Worker** + * **Backend -> Worker** ```json { "type": "unsubscribe", @@ -437,7 +482,7 @@ This section shows a typical sequence of messages between the backend and the wo } ``` 6. **Worker Acknowledges Unsubscription** - - **Worker -> Backend** + * **Worker -> Backend** ```json { "type": "stateReport", @@ -448,7 +493,6 @@ This section shows a typical sequence of messages between the backend and the wo "cameraConnections": [] } ``` - ## 7. HTTP API: Image Retrieval In addition to the WebSocket protocol, the worker exposes an HTTP endpoint for retrieving the latest image frame from a camera. @@ -464,13 +508,11 @@ GET /camera/{camera_id}/image ### Response - **Success (200):** Returns the latest JPEG image from the camera stream. - - - `Content-Type: image/jpeg` - - Binary JPEG data. + - `Content-Type: image/jpeg` + - Binary JPEG data. - **Error (404):** If the camera is not found or no frame is available. - - - JSON error response. + - JSON error response. - **Error (500):** Internal server error. @@ -483,9 +525,9 @@ GET /camera/display-001;cam-001/image ### Example Response - **Headers:** - ``` - Content-Type: image/jpeg - ``` + ``` + Content-Type: image/jpeg + ``` - **Body:** Binary JPEG image. ### Notes