From fd1eeeda47d8d93532aa8d023ebb96e31029e0aa Mon Sep 17 00:00:00 2001 From: Cobra Date: Fri, 10 Apr 2026 15:02:26 -0400 Subject: [PATCH] Fix P1 blockers: configurable port, connection leak, race condition, DB path + WAL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P1 #551: Add STATUS_PORT env var (default 9191) for configurable HTTP status server port - P1 #552: Fix SQLite connection leak — wrap all sqlite3.connect() with try/finally in lookup_person(), seed_persons_from_db(), log_signal(), log_presence_change() - P1 #553: Fix race condition in update_person_state() — compute certainty snapshot under devices_lock, determine state transition under persons_lock, send alert outside locks to prevent stale snapshot - P2 #559 (included): Add WAL mode + pragmas to init_db() for concurrent write safety - Database path: Align to /opt/sensor/presence.db in both daemon and install.sh - Increase connection timeout from 1s to 10s across all functions - install.sh: Document PRESENCE_STATUS_PORT and PRESENCE_MATRIX_WEBHOOK env vars --- presence/install.sh | 6 ++- presence/presence_daemon.py | 73 +++++++++++++++++++++++-------------- 2 files changed, 51 insertions(+), 28 deletions(-) diff --git a/presence/install.sh b/presence/install.sh index ef61fba..c197b3a 100644 --- a/presence/install.sh +++ b/presence/install.sh @@ -28,7 +28,11 @@ Type=simple ExecStart=/usr/bin/python3 /opt/sensor/sensor.py Restart=on-failure RestartSec=10 -Environment=PRESENCE_DB_PATH=/opt/sensor/sensor.db +Environment=PRESENCE_DB_PATH=/opt/sensor/presence.db +# Optional: override default status port (9191) +# Environment=PRESENCE_STATUS_PORT=9191 +# Optional: configure Matrix webhook for alerts +# Environment=PRESENCE_MATRIX_WEBHOOK=https://matrix.example.com/hook [Install] WantedBy=multi-user.target diff --git a/presence/presence_daemon.py b/presence/presence_daemon.py index 419df9a..61c40d1 100644 --- a/presence/presence_daemon.py +++ b/presence/presence_daemon.py @@ -26,8 +26,9 @@ from typing import Optional # --- Config (env-driven) --- MONITOR_IFACE = os.getenv("PRESENCE_MONITOR_IFACE", "wlan1") DATA_IFACE = os.getenv("PRESENCE_DATA_IFACE", "wlan0") -DB_PATH = os.getenv("PRESENCE_DB_PATH", "/opt/presence/presence.db") +DB_PATH = os.getenv("PRESENCE_DB_PATH", "/opt/sensor/presence.db") MATRIX_WEBHOOK = os.getenv("PRESENCE_MATRIX_WEBHOOK", "") +STATUS_PORT = int(os.getenv("PRESENCE_STATUS_PORT", "9191")) PRESENCE_THRESHOLD = 0.40 ABSENCE_THRESHOLD = 0.10 ANCHOR_MINUTES = 45 @@ -138,42 +139,47 @@ def init_db() -> None: db_path = Path(DB_PATH) db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(DB_PATH, timeout=10) try: - conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() schema = schema_path.read_text() cursor.executescript(schema) conn.commit() - conn.close() + cursor.execute("PRAGMA journal_mode=WAL") + cursor.execute("PRAGMA synchronous=NORMAL") + cursor.execute("PRAGMA foreign_keys=ON") + conn.commit() logger.info(f"Database initialized at {DB_PATH}") except Exception as e: logger.error(f"Failed to init database: {e}") + finally: + conn.close() def lookup_person(mac: str) -> Optional[int]: """Look up person_id for a MAC from database.""" + conn = sqlite3.connect(DB_PATH, timeout=10) try: - conn = sqlite3.connect(DB_PATH, timeout=1) cursor = conn.cursor() result = cursor.execute( "SELECT person_id FROM devices WHERE mac = ?", (mac.lower(),) ).fetchone() - conn.close() return result[0] if result else None except Exception as e: logger.debug(f"lookup_person error for {mac}: {e}") return None + finally: + conn.close() def seed_persons_from_db() -> None: """Load all persons from database into memory.""" + conn = sqlite3.connect(DB_PATH, timeout=10) try: - conn = sqlite3.connect(DB_PATH, timeout=1) conn.row_factory = sqlite3.Row cursor = conn.cursor() rows = cursor.execute("SELECT id, name FROM persons").fetchall() - conn.close() with persons_lock: for row in rows: @@ -185,36 +191,40 @@ def seed_persons_from_db() -> None: logger.info(f"Loaded {len(rows)} persons from database") except Exception as e: logger.error(f"Failed to seed persons: {e}") + finally: + conn.close() def log_signal(mac: str, signal_type: str, certainty: float) -> None: """Insert signal into database (non-blocking via queue would be ideal, but inline for simplicity).""" + conn = sqlite3.connect(DB_PATH, timeout=10) try: - conn = sqlite3.connect(DB_PATH, timeout=1) cursor = conn.cursor() cursor.execute( "INSERT INTO signals (mac, signal_type, certainty, ts) VALUES (?, ?, ?, ?)", (mac.lower(), signal_type, certainty, time.time()) ) conn.commit() - conn.close() except Exception as e: logger.debug(f"Failed to log signal: {e}") + finally: + conn.close() def log_presence_change(person_id: int, old_status: str, new_status: str) -> None: """Insert occupancy change into database.""" + conn = sqlite3.connect(DB_PATH, timeout=10) try: - conn = sqlite3.connect(DB_PATH, timeout=1) cursor = conn.cursor() cursor.execute( "INSERT INTO occupancy_log (person_id, old_status, new_status, ts) VALUES (?, ?, ?, ?)", (person_id, old_status, new_status, time.time()) ) conn.commit() - conn.close() except Exception as e: logger.debug(f"Failed to log presence change: {e}") + finally: + conn.close() def send_alert(person_name: str, status: str, signal_type: str, certainty: float) -> None: @@ -277,6 +287,7 @@ def update_person_state(person_id: Optional[int]) -> None: if person_id is None: return + # Snapshot device state under lock with devices_lock: # Get all devices for this person devices_for_person = [d for d in devices.values() if d.person_id == person_id] @@ -293,34 +304,42 @@ def update_person_state(person_id: Optional[int]) -> None: # Get max last_seen time max_last_seen = max(d.last_seen for d in devices_for_person) + # Determine new status based on snapshot + current_time = time.time() + elapsed_since_signal = (current_time - max_last_seen) / 60.0 + with persons_lock: person = persons.get(person_id) if person is None: return old_status = person.status - elapsed_since_signal = (time.time() - max_last_seen) / 60.0 + new_status = old_status # State machine logic if person.status == "UNKNOWN": if agg_certainty >= PRESENCE_THRESHOLD: - person.status = "PRESENT" - person.last_change = time.time() - log_presence_change(person_id, old_status, person.status) - send_alert(person.name, "PRESENT", "signal", agg_certainty) + new_status = "PRESENT" elif person.status == "PRESENT": # Check presence anchor if elapsed_since_signal >= ANCHOR_MINUTES: - person.status = "ABSENT" - person.last_change = time.time() - log_presence_change(person_id, old_status, person.status) - send_alert(person.name, "ABSENT", "timeout", agg_certainty) + new_status = "ABSENT" elif person.status == "ABSENT": if agg_certainty >= PRESENCE_THRESHOLD: - person.status = "PRESENT" - person.last_change = time.time() - log_presence_change(person_id, old_status, person.status) - send_alert(person.name, "PRESENT", "signal", agg_certainty) + new_status = "PRESENT" + + # Apply state transition if changed + if new_status != old_status: + person.status = new_status + person.last_change = current_time + log_presence_change(person_id, old_status, new_status) + + # Send alert outside of lock + if new_status != old_status: + if new_status == "PRESENT": + send_alert(person.name, "PRESENT", "signal", agg_certainty) + else: + send_alert(person.name, "ABSENT", "timeout", agg_certainty) def probe_sniffer(iface: str) -> None: @@ -646,13 +665,13 @@ def status_handler(request, client_address, server): def status_server() -> None: - """Start HTTP status server on 127.0.0.1:9191.""" + """Start HTTP status server on 127.0.0.1:STATUS_PORT.""" try: server = http.server.HTTPServer( - ("127.0.0.1", 9191), + ("127.0.0.1", STATUS_PORT), status_handler ) - logger.info("Status server started on http://127.0.0.1:9191") + logger.info(f"Status server started on http://127.0.0.1:{STATUS_PORT}") server.serve_forever() except Exception as e: logger.error(f"Failed to start status server: {e}")