Fix MQTT cleanup lifetime race, ONVIF concurrency/parsing safety, and MQTT docs payload mismatch (#452)

* Initial plan

* Fix MQTT and ONVIF safety issues with doc sync

* Harden MQTT cleanup timeout synchronization

* Address review nits in ONVIF and MQTT timeout logging

* Handle MQTT timeout clock and log edge cases safely

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>

Author: Copilot

docs/MQTT_INTEGRATION.md

@@ -119,7 +119,6 @@ Detection events are published as JSON with the following structure:
 {
   "stream": "front_door",
   "timestamp": 1706745600,
-  "timestamp_iso": "2024-02-01T12:00:00Z",
   "count": 2,
   "snapshot_path": "/var/lib/lightnvr/recordings/snapshots/front_door/20240201_120000.jpg",
   "snapshot_url": "/api/snapshots/front_door/20240201_120000.jpg",
@@ -154,7 +153,6 @@ Detection events are published as JSON with the following structure:
 |-------|-------------|
 | `stream` | Name of the camera/stream |
 | `timestamp` | Unix timestamp (seconds since epoch) |
-| `timestamp_iso` | ISO 8601 formatted timestamp |
 | `count` | Number of detections in this event |
 | `snapshot_path` | Local filesystem path of the event snapshot (omitted if capture failed) |
 | `snapshot_url` | Relative URL to fetch the snapshot from the LightNVR web server (uses normal authentication) |
@@ -289,7 +287,7 @@ def on_connect(client, userdata, flags, rc):
 def on_message(client, userdata, msg):
     data = json.loads(msg.payload)
     stream = data['stream']
-    timestamp = data['timestamp_iso']
+    timestamp = data['timestamp']
 
     for det in data['detections']:
         print(f"[{timestamp}] {stream}: {det['label']} ({det['confidence']:.0%})")
@@ -430,4 +428,3 @@ client_id = lightnvr-garage
 client_id = lightnvr-frontyard
 ```
 
-

src/core/mqtt_client.c

@@ -1221,11 +1221,14 @@ typedef enum {
     MQTT_OP_LIB_CLEANUP
 } mqtt_cleanup_op_t;
 
-// Thread argument structure - uses a simple volatile flag for completion
+// Thread argument structure for timeout-aware cleanup worker
 typedef struct {
     struct mosquitto *mosq;
     mqtt_cleanup_op_t op;
-    volatile int *done_flag;  // Pointer to completion flag
+    pthread_mutex_t mutex;
+    pthread_cond_t cond;
+    bool done;
+    bool timed_out;
 } mqtt_cleanup_arg_t;
 
 /**
@@ -1234,7 +1237,6 @@ typedef struct {
 static void *mqtt_cleanup_thread(void *arg) {
     log_set_thread_context("MQTT", NULL);
     mqtt_cleanup_arg_t *cleanup_arg = (mqtt_cleanup_arg_t *)arg;
-    volatile int *done_flag = cleanup_arg->done_flag;
 
     switch (cleanup_arg->op) {
         case MQTT_OP_LOOP_STOP:
@@ -1248,58 +1250,105 @@ static void *mqtt_cleanup_thread(void *arg) {
             break;
     }
 
-    // Signal completion by setting the flag
-    // Note: cleanup_arg is on the stack of the caller and remains valid
-    // until the caller returns (after timeout or completion)
-    *done_flag = 1;
+    pthread_mutex_lock(&cleanup_arg->mutex);
+    cleanup_arg->done = true;
+    pthread_cond_signal(&cleanup_arg->cond);
+    bool timed_out = cleanup_arg->timed_out;
+    pthread_mutex_unlock(&cleanup_arg->mutex);
+
+    // On timeout, caller detached and returned; worker owns cleanup.
+    if (timed_out) {
+        pthread_cond_destroy(&cleanup_arg->cond);
+        pthread_mutex_destroy(&cleanup_arg->mutex);
+        free(cleanup_arg);
+    }
 
     return NULL;
 }
 
 /**
  * Run a mosquitto cleanup operation with a timeout
- * Uses simple polling with usleep - portable across glibc and musl
+ * Uses a timed condition wait with a worker thread
  * Returns true if completed within timeout, false if timed out
  */
 static bool mqtt_run_with_timeout(struct mosquitto *m, mqtt_cleanup_op_t op, int timeout_sec, const char *op_name) {
     pthread_t thread;
-    volatile int done_flag = 0;
+    mqtt_cleanup_arg_t *arg = (mqtt_cleanup_arg_t *)calloc(1, sizeof(mqtt_cleanup_arg_t));
 
-    // Argument structure on stack - valid for duration of this function
-    mqtt_cleanup_arg_t arg;
-    arg.mosq = m;
-    arg.op = op;
-    arg.done_flag = &done_flag;
+    if (arg == NULL) {
+        log_warn("MQTT: Failed to allocate resources for %s", op_name);
+        return false;
+    }
+
+    if (pthread_mutex_init(&arg->mutex, NULL) != 0) {
+        free(arg);
+        log_warn("MQTT: Failed to initialize synchronization objects for %s", op_name);
+        return false;
+    }
+    if (pthread_cond_init(&arg->cond, NULL) != 0) {
+        pthread_mutex_destroy(&arg->mutex);
+        free(arg);
+        log_warn("MQTT: Failed to initialize synchronization objects for %s", op_name);
+        return false;
+    }
+
+    arg->mosq = m;
+    arg->op = op;
 
     // Create thread to run the operation
-    if (pthread_create(&thread, NULL, mqtt_cleanup_thread, &arg) != 0) {
+    if (pthread_create(&thread, NULL, mqtt_cleanup_thread, arg) != 0) {
         log_warn("MQTT: Failed to create thread for %s", op_name);
+        pthread_cond_destroy(&arg->cond);
+        pthread_mutex_destroy(&arg->mutex);
+        free(arg);
         return false;
     }
 
-    // Detach the thread so it cleans up automatically if we timeout
-    pthread_detach(thread);
+    struct timespec ts;
+    if (clock_gettime(CLOCK_REALTIME, &ts) != 0) {
+        log_warn("MQTT: Failed to get clock time for %s timeout", op_name);
+        pthread_mutex_lock(&arg->mutex);
+        arg->timed_out = true;
+        pthread_mutex_unlock(&arg->mutex);
+        pthread_detach(thread);
+        return false;
+    }
+    ts.tv_sec += timeout_sec;
 
-    // Poll for completion with 50ms intervals
-    int timeout_ms = timeout_sec * 1000;
-    int elapsed_ms = 0;
-    const int poll_interval_ms = 50;
+    pthread_mutex_lock(&arg->mutex);
+    int wait_rc = 0;
+    while (!arg->done && wait_rc != ETIMEDOUT) {
+        wait_rc = pthread_cond_timedwait(&arg->cond, &arg->mutex, &ts);
+    }
+    bool completed = arg->done;
+    if (!completed) {
+        arg->timed_out = true;
+    }
+    pthread_mutex_unlock(&arg->mutex);
 
-    while (elapsed_ms < timeout_ms) {
-        if (done_flag) {
-            // Operation completed successfully
-            return true;
+    if (completed) {
+        // Operation completed successfully; caller reclaims resources.
+        if (pthread_join(thread, NULL) != 0) {
+            log_warn("MQTT: Failed to join thread for %s", op_name);
         }
-        usleep(poll_interval_ms * 1000);
-        elapsed_ms += poll_interval_ms;
+        pthread_cond_destroy(&arg->cond);
+        pthread_mutex_destroy(&arg->mutex);
+        free(arg);
+        return true;
     }
 
-    // Timeout - the thread is still running but we'll return anyway
-    // The thread will eventually complete (or process will exit)
+    // Timeout - detach so worker can finish and free its resources.
+    pthread_detach(thread);
+
     // Use write() first to ensure we see the timeout even if logger is blocked
     char timeout_msg[128];
-    snprintf(timeout_msg, sizeof(timeout_msg), "[MQTT DEBUG] %s timed out after %d seconds\n", op_name, timeout_sec);
-    (void)write(STDERR_FILENO, timeout_msg, strlen(timeout_msg));
+    int written = snprintf(timeout_msg, sizeof(timeout_msg), "[MQTT DEBUG] %s timed out after %d seconds\n", op_name, timeout_sec);
+    if (written >= 0) {
+        size_t msg_len = ((size_t)written < sizeof(timeout_msg)) ? (size_t)written : sizeof(timeout_msg) - 1;
+        if (msg_len > 0) {
+            (void)write(STDERR_FILENO, timeout_msg, msg_len);
+        }
+    }
     log_warn("MQTT: %s timed out after %d seconds", op_name, timeout_sec);
     return false;
 }
@@ -1493,4 +1542,3 @@ int mqtt_reinit(const config_t *config) {
 }
 
 #endif /* ENABLE_MQTT */
-

src/video/onvif_detection.c

@@ -248,7 +248,11 @@ static char *send_onvif_request(const char *url, const char *username, const cha
 
     // Create full URL
     char full_url[512];
-    snprintf(full_url, sizeof(full_url), "%s/onvif/%s", url, service);
+    int written = snprintf(full_url, sizeof(full_url), "%s/onvif/%s", url, service);
+    if (written < 0 || (size_t)written >= sizeof(full_url)) {
+        log_error("ONVIF request URL too long or formatting failed");
+        return NULL;
+    }
 
     return send_onvif_request_to_url(full_url, username, password, request_body, action);
 }
@@ -265,14 +269,19 @@ static char *extract_subscription_address(const char *response) {
     };
 
     for (int i = 0; i < 3; i++) {
-        const char *start = strstr(response, patterns[(ptrdiff_t)i*2]);
-        const char *end = strstr(response, patterns[(ptrdiff_t)i*2+1]);
-        
-        if (start && end) {
-            start += strlen(patterns[(ptrdiff_t)i * 2]);
-            int length = (int)(end - start);
-            
-            return strndup(start, length);
+        const char *open_tag = patterns[(ptrdiff_t)i * 2];
+        const char *close_tag = patterns[(ptrdiff_t)i * 2 + 1];
+        const char *start = strstr(response, open_tag);
+
+        if (start) {
+            size_t open_tag_len = strlen(open_tag);
+            const char *content_start = start + open_tag_len;
+            const char *end = strstr(content_start, close_tag);
+
+            if (end) {
+                size_t length = (size_t)(end - content_start);
+                return strndup(content_start, length);
+            }
         }
     }
 
@@ -608,13 +617,14 @@ static bool has_motion_event(const char *response) {
  * Initialize the ONVIF detection system
  */
 int init_onvif_detection_system(void) {
+    pthread_mutex_lock(&curl_mutex);
+
     if (initialized && curl_handle) {
         log_info("ONVIF detection system already initialized");
-        return 0;  // Already initialized and curl handle is valid
+        pthread_mutex_unlock(&curl_mutex);
+        return 0;  // Already initialized and curl handle is present
     }
 
-    pthread_mutex_lock(&curl_mutex);
-
     // If we have a curl handle but initialized is false, clean it up first
     if (curl_handle) {
         log_warn("ONVIF detection system has a curl handle but is marked as uninitialized, cleaning up");
@@ -637,15 +647,14 @@ int init_onvif_detection_system(void) {
         return -1;
     }
 
-    pthread_mutex_unlock(&curl_mutex);
-
     // Initialize subscriptions
     pthread_mutex_lock(&subscription_mutex);
     subscription_count = 0;
     memset(subscriptions, 0, sizeof(subscriptions));
     pthread_mutex_unlock(&subscription_mutex);
 
     initialized = true;
+    pthread_mutex_unlock(&curl_mutex);
     log_info("ONVIF detection system initialized successfully");
     return 0;
 }
@@ -654,22 +663,22 @@ int init_onvif_detection_system(void) {
  * Shutdown the ONVIF detection system
  */
 void shutdown_onvif_detection_system(void) {
+    pthread_mutex_lock(&curl_mutex);
     log_info("Shutting down ONVIF detection system (initialized: %s, curl_handle: %p)",
              initialized ? "yes" : "no", (void*)curl_handle);
 
     // Cleanup curl handle if it exists
-    pthread_mutex_lock(&curl_mutex);
     if (curl_handle) {
         log_info("Cleaning up curl handle");
         curl_easy_cleanup(curl_handle);
         curl_handle = NULL;
     }
-    pthread_mutex_unlock(&curl_mutex);
 
     // Note: Don't call curl_global_cleanup() here - it's managed centrally in curl_init.c
     // The global cleanup will happen at program shutdown
 
     initialized = false;
+    pthread_mutex_unlock(&curl_mutex);
     log_info("ONVIF detection system shutdown complete");
 }
 
@@ -763,13 +772,18 @@ int detect_motion_onvif(const char *onvif_url, const char *username, const char
 
     if (!response) {
         log_error("Failed to pull messages from subscription");
-        
-        // If pulling messages fails, the subscription might be invalid
-        // Mark it as inactive so we'll create a new one next time
+
+        // If pulling messages fails, the subscription might be invalid.
+        // Re-lookup by URL while holding the mutex to avoid stale pointer use.
         pthread_mutex_lock(&subscription_mutex);
-        subscription->active = false;
+        for (int i = 0; i < subscription_count; i++) {
+            if (strcmp(subscriptions[i].camera_url, onvif_url) == 0) {
+                subscriptions[i].active = false;
+                break;
+            }
+        }
         pthread_mutex_unlock(&subscription_mutex);
-        
+
         return -1;
     }