Preserve stream control events when discarding delayed media

tests/test_android_stream.c

@@ -57,10 +57,16 @@ int main(void) {
     CHECK(android_stream_feed(1, 1, 100000, 0, video, sizeof(video)) == 0);
     CHECK(android_stream_feed(1, 1, 100000, 1, video, sizeof(video)) == 1);
     CHECK(android_stream_feed(1, 0, 100000, 0, (const uint8_t *)"ops", 3) == 1);
-    uint8_t *output = wait_event(2, 1, &size);
-    CHECK(size == sizeof(video) + 20 && stream_read_be(output + 8, 8) == 100000 && !memcmp(output + 20, video, sizeof(video)));
-    u_free(output);
-    output = wait_event(2, 2, &size); CHECK(size == 23 && !memcmp(output + 20, "ops", 3)); u_free(output);
+    uint8_t *output = NULL; int video_seen = 0, audio_seen = 0;
+    for (unsigned i = 0; i < 1000 && (!video_seen || !audio_seen); i++) {
+        output = android_stream_pull(2, &size);
+        if (output && output[0] == 1) {
+            CHECK(size == sizeof(video) + 20 && stream_read_be(output + 8, 8) == 100000 && !memcmp(output + 20, video, sizeof(video)));
+            video_seen++;
+        } else if (output && output[0] == 2) { CHECK(size == 23 && !memcmp(output + 20, "ops", 3)); audio_seen++; }
+        u_free(output); test_clock_poll(ua, 100);
+    }
+    CHECK(video_seen == 1 && audio_seen == 1);
     android_stream_stop(1); u_free(wait_event(2, 5, &size)); android_stream_stop(2); u_free(wait_event(1, 5, &size));
     while ((output = android_stream_pull(2, &size))) u_free(output);
     CHECK(android_stream_open(3, group, 0, 1280, 720, 0, 0) == 0); u_free(wait_event(3, 4, &size));

tools/chatgui-android/libutun_lite/stream_bridge.c

@@ -65,6 +65,18 @@ static void queue_clear(struct media_queue *q) {
     while ((record = queue_take(q))) u_free(record);
 }
 
+/* RX pressure удаляет media/GAP, сохраняя подтверждения открытия/закрытия owners. */
+static void media_drop_output(void) {
+    struct media_record **link = &output_queue.head, *previous = NULL;
+    while (*link) {
+        struct media_record *r = *link;
+        if (r->data[0] <= 3 || (r->id != publisher.id && r->id != viewer.id)) {
+            *link = r->next; if (output_queue.tail == r) output_queue.tail = previous;
+            output_queue.bytes -= r->length; output_queue.count--; u_free(r);
+        } else { previous = r; link = &r->next; }
+    }
+}
+
 /* Все обращения к ua, включая post, защищены тем же mutex, что и detach перед destroy. */
 static int media_schedule(void) {
     if (media_post || media_timer) return 0;
@@ -86,7 +98,7 @@ static void media_output(uint64_t id, int kind, const struct stream_media_format
         DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream RX full id=%llu bytes=%zu count=%u",
                    (unsigned long long)id, output_queue.bytes, output_queue.count);
         viewer.rx_overflow = 1;
-        queue_clear(&output_queue);
+        media_drop_output();
         if (kind == 1 || kind == 2) { media_output(id, 3, NULL, 0, 0, NULL, 0); return; }
     }
     struct media_record *record = u_calloc(1, sizeof(*record) + size + 20);
@@ -190,10 +202,14 @@ static void media_list_sources(void) {
     int failed = strbuf_addf(&json, "[") < 0;
     if (group && !failed) {
         char query[256];
+        for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
+            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source namespace invalid group=%llx", (unsigned long long)group_id);
+            failed = 1; break;
+        }
         snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') "
                  "FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id", group->channel_id);
         sqlite3 *db = chat_core_get_db(media_inst);
-        int result = sqlite3_prepare_v2(db, query, -1, &statement, NULL), count = 0;
+        int result = failed ? SQLITE_ERROR : sqlite3_prepare_v2(db, query, -1, &statement, NULL), count = 0;
         if (result == SQLITE_OK) {
             while ((result = sqlite3_step(statement)) == SQLITE_ROW) {
                 uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0);