Exercise video switching and backpressure in a real call

src/call/call_video_transport.c

@@ -76,7 +76,8 @@ static void flush(struct call_video_transport *v) {
     uint8_t wire[CALL_VIDEO_HEADER_BYTES + CALL_VIDEO_FRAGMENT_BYTES];
     // Ограниченный burst одного вызова; повторяется по audio send и таймеру 50мс.
     for (unsigned budget = 0; v->data && budget < 16; budget++) {
-        if (!etcp_router_send_q_has_room(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO)) return;
+        if (!etcp_router_send_q_has_room(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO)
+            || queue_entry_count(conn->send_q) >= ROUTER_MAX_SEND_Q_PACKETS) return;
         uint32_t remaining = v->tx.size - v->tx.offset;
         v->tx.length = remaining < CALL_VIDEO_FRAGMENT_BYTES ? remaining : CALL_VIDEO_FRAGMENT_BYTES;
         v->tx.data = v->data + v->tx.offset;

tests/test_call.c

@@ -17,6 +17,7 @@
 #include "../dm/dm_core.h"
 #include "../dm/dm_crypto.h"
 #include "../call/call_proto.h"
+#include "../call/call_video_transport.h"
 #include "../routing_layer/topo_node_sqlite.h"
 #include "../routing_layer/topo_group.h"
 #include "../routing_layer/topo_node.h"
@@ -60,6 +61,7 @@ enum call_phase {
     P_WAIT_ACCEPTED,     /* обе ACTIVE */
     P_SEND_MEDIA,        /* A: call_send_media (тестовый Opus-кадр) */
     P_WAIT_MEDIA,        /* B: media_recv_cb получил кадр */
+    P_VIDEO_CONFIG, P_VIDEO_SEND, P_VIDEO_HOLD, P_VIDEO_BUSY, P_VIDEO_RESUME, P_VIDEO_SWITCH, P_VIDEO_SWITCH_WAIT,
     P_HANGUP,            /* A: call_hangup */
     P_WAIT_ENDED1,       /* обе ENDED */
     P_CALL2_START,       /* A: call_start (2-й звонок) */
@@ -75,6 +77,7 @@ struct call_ctx {
     enum call_phase phase;
     int   result;        /* 0=running, 1=ok, 2=fail */
     int   ticks;
+    int phase_ticks;
     void* timer;
     uint64_t call_id;    /* текущий call_id */
 };
@@ -109,6 +112,27 @@ static void audio_recv_cb(struct UTUN_INSTANCE* inst, uint64_t call_id, uint16_t
 }
 static const struct call_audio_ops audio_ops = {.on_media = audio_recv_cb};
 
+static struct call_video_caps g_video_peer[2];
+static uint8_t g_video_data[9001];
+static int g_video_frames, g_video_ready, g_video_error;
+static void video_caps(void *arg, const struct call_video_caps *caps) { g_video_peer[(size_t)arg] = *caps; }
+static int video_ready(void *arg, uint32_t bytes) { (void)arg; return bytes <= sizeof(g_video_data) && g_video_ready; }
+static void video_frame(void *arg, const struct call_video_frame *f) {
+    (void)arg;
+    g_video_frames++;
+    if (f->size != sizeof(g_video_data) || memcmp(f->data, g_video_data, sizeof(g_video_data))
+        || f->format.generation != (uint64_t)g_video_frames || f->format.codec != (g_video_frames == 1 ? CALL_VIDEO_H265 : CALL_VIDEO_H264))
+        g_video_error = 1;
+}
+static const struct call_video_ops video_ops = {.capabilities = video_caps, .receive_ready = video_ready, .frame = video_frame};
+static struct call_video_frame video_test_frame(uint64_t id, unsigned generation) {
+    struct call_video_frame f = {0};
+    f.call_id = id; f.frame_id = generation; f.pts_us = sent_timing.pts_us + generation * 30000;
+    f.format = (struct call_video_format){generation, 640, 480, 30, 0, generation == 1 ? CALL_VIDEO_H265 : CALL_VIDEO_H264};
+    f.flags = CALL_VIDEO_KEY; f.size = f.length = sizeof(g_video_data); f.data = g_video_data;
+    return f;
+}
+
 static void media_recv_cb(struct UTUN_INSTANCE* inst, uint64_t call_id,
                           const uint8_t* opus, int len, void* arg) {
     (void)inst;
@@ -461,9 +485,59 @@ static void call_tick(void* arg) {
                 || g_capture_timing.epoch != sent_timing.epoch || g_capture_timing.delay_samples != sent_timing.delay_samples) {
                 fprintf(stderr, "B: capture timing changed in transit\n"); t->result = 2; break;
             }
-            t->phase = P_HANGUP;
+            t->phase = P_VIDEO_CONFIG;
         }
         break;
+    case P_VIDEO_CONFIG: {
+        for (unsigned i = 4; i < sizeof(g_video_data); i++) g_video_data[i] = (uint8_t)i;
+        struct call_video_caps caps = {0};
+        caps.encode[0] = caps.encode[1] = caps.decode[0] = caps.decode[1] = (struct call_video_limit){1280, 720, 30, 1280 * 720 * 30};
+        if (call_video_configure(A, t->call_id, &caps, &video_ops, (void*)0)
+            || call_video_configure(B, t->call_id, &caps, &video_ops, (void*)1)) { t->result = 2; break; }
+        g_video_ready = 0; t->phase = P_VIDEO_SEND; break;
+    }
+    case P_VIDEO_SEND: {
+        if (g_video_peer[0].decode[1].width != 1280 || g_video_peer[1].decode[1].width != 1280) break;
+        struct call_video_frame f = video_test_frame(t->call_id, 1);
+        int result = call_video_send_frame(A, &f);
+        if (result < 0) { fprintf(stderr, "video send failed\n"); t->result = 2; break; }
+        if (result == 1) { t->phase_ticks = 0; t->phase = P_VIDEO_HOLD; }
+        break;
+    }
+    case P_VIDEO_HOLD: {
+        if (g_video_frames) { fprintf(stderr, "video escaped audio barrier\n"); t->result = 2; break; }
+        if (++t->phase_ticks < 10) break;
+        struct call_capture_time timing = sent_timing; timing.pts_us += 20000;
+        uint8_t opus[16] = {1};
+        if (call_send_media(A, t->call_id, opus, sizeof(opus), &timing)) { t->result = 2; break; }
+        t->phase_ticks = 0; t->phase = P_VIDEO_BUSY; break;
+    }
+    case P_VIDEO_BUSY:
+        if (g_video_frames) { fprintf(stderr, "video bypassed receive backpressure\n"); t->result = 2; break; }
+        if (++t->phase_ticks < 10 || g_media[IDX_B].frames != 2) break;
+        g_video_ready = 1; call_video_resume_receive(B, t->call_id); t->phase = P_VIDEO_RESUME; break;
+    case P_VIDEO_RESUME:
+        if (!g_video_frames) break;
+        if (g_video_error) { t->result = 2; break; }
+        t->phase = P_VIDEO_SWITCH; break;
+    case P_VIDEO_SWITCH: {
+        struct call_video_frame f = video_test_frame(t->call_id, 2);
+        if (call_video_send_frame(A, &f) != 1) { t->result = 2; break; }
+        struct call_capture_time timing = sent_timing; timing.pts_us += 60000;
+        uint8_t opus[16] = {2};
+        if (call_send_media(A, t->call_id, opus, sizeof(opus), &timing)) { t->result = 2; break; }
+        t->phase = P_VIDEO_SWITCH_WAIT; break;
+    }
+    case P_VIDEO_SWITCH_WAIT:
+        if (g_video_frames < 2) break;
+        if (g_video_error) { t->result = 2; break; }
+        // Завершение во время незавершённой сборки/паузы приёма не оставляет callbacks.
+        g_video_ready = 0;
+        {
+            struct call_video_frame f = video_test_frame(t->call_id, 3);
+            if (call_video_send_frame(A, &f) != 1) { t->result = 2; break; }
+        }
+        t->phase = P_HANGUP; break;
     case P_HANGUP:
         call_hangup(A, t->call_id);
         t->phase = P_WAIT_ENDED1;