Bound video encoder commands and verify generation-safe feedback

src/call/call_video_transport.c

@@ -24,7 +24,7 @@ struct call_video_transport {
     int have_peer, need_key;
     struct call_video_rate rate;
     struct route_tx_meter production;
-    uint64_t rate_generation, rate_request, produced_id, route_local_epoch, route_peer_epoch;
+    uint64_t rate_generation, rate_request, produced_id, route_local_epoch, route_peer_epoch, rate_requested_us;
     uint32_t applied_bps, maximum_bps;
     int rate_pending, admit;
 };
@@ -80,6 +80,12 @@ static void update_rate(struct call_video_transport *v) {
     if (!v->rate_generation || !v->applied_bps) return;
     struct call_session *s = v->session;
     uint64_t now = get_time_us();
+    if (v->rate_pending && now - v->rate_requested_us > 2000000) {
+        v->rate_pending = 0; v->applied_bps = 0; v->admit = 0;
+        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video encoder command timeout call=%llu generation=%llu request=%llu",
+                    (unsigned long long)s->call_id, (unsigned long long)v->rate_generation, (unsigned long long)v->rate_request);
+        return;
+    }
     struct ETCP_ROUTER_TX_STATS net = {0};
     int found = etcp_router_tx_stats(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO, now, &net) == 0;
     if (found && (net.local_epoch != v->route_local_epoch || net.peer_epoch != v->route_peer_epoch)) {
@@ -109,7 +115,7 @@ static void update_rate(struct call_video_transport *v) {
     if (call_video_rate_step(&v->rate, &sample, &result)) { v->admit = 0; return; }
     v->admit = result.admit;
     if (result.target_bps != v->applied_bps && v->ops.bitrate) {
-        v->rate_pending = 1; ++v->rate_request;
+        v->rate_pending = 1; ++v->rate_request; v->rate_requested_us = now;
         v->ops.bitrate(v->arg, v->rate_generation, v->rate_request, result.target_bps);
     }
 }
@@ -204,6 +210,11 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr
         if (!v->admit) { v->need_key = 1; return 0; }
     }
     if (v->data) { v->need_key = 1; return 0; }
+    struct ETCP_ROUTER_TX_STATS net;
+    if (etcp_router_tx_stats(inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO, get_time_us(), &net) == 0) {
+        uint64_t debt = media_bytes(net.queued_bytes + net.inflight_bytes, (uint64_t)net.queued_packets + net.inflight_packets);
+        if (debt + f->size > CALL_VIDEO_MAX_FRAME) { v->need_key = 1; return 0; }
+    }
     int changed = !call_video_same_format(&f->format, &v->tx.format);
     if (f->format.generation < v->tx.format.generation || (changed && f->format.generation == v->tx.format.generation)) {
         DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX format changed without new generation"); return -1;

tests/test_call.c

@@ -115,6 +115,9 @@ 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 uint64_t g_rate_generation, g_rate_request;
+static uint32_t g_rate_bps;
+static unsigned g_rate_commands;
 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) {
@@ -124,7 +127,12 @@ static void video_frame(void *arg, const struct call_video_frame *f) {
         || 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 void video_bitrate(void *arg, uint64_t generation, uint64_t request, uint32_t bps) {
+    (void)arg; g_rate_generation = generation; g_rate_request = request; g_rate_bps = bps; g_rate_commands++;
+}
+static const struct call_video_ops video_ops = {
+    .capabilities = video_caps, .receive_ready = video_ready, .frame = video_frame, .bitrate = video_bitrate
+};
 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;
@@ -494,6 +502,7 @@ static void call_tick(void* arg) {
         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; }
+        if (call_video_configure_rate(A, t->call_id, 1, 800000, 400000)) { t->result = 2; break; }
         g_video_ready = 0; t->phase = P_VIDEO_SEND; break;
     }
     case P_VIDEO_SEND: {
@@ -517,10 +526,18 @@ static void call_tick(void* arg) {
         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_frames || !g_rate_commands) break;
         if (g_video_error) { t->result = 2; break; }
+        if (g_rate_commands != 1 || g_rate_generation != 1 || g_rate_bps != 400000
+            || call_video_rate_applied(A, t->call_id, 2, g_rate_request, g_rate_bps) != -1
+            || call_video_rate_applied(A, t->call_id, 1, g_rate_request + 1, g_rate_bps) != -1
+            || call_video_rate_applied(A, t->call_id, 1, g_rate_request, g_rate_bps)) {
+            fprintf(stderr, "video rate command/completion mismatch\n"); t->result = 2; break;
+        }
         t->phase = P_VIDEO_SWITCH; break;
     case P_VIDEO_SWITCH: {
+        if (call_video_configure_rate(A, t->call_id, 2, 400000, 400000)
+            || call_video_rate_applied(A, t->call_id, 1, g_rate_request, g_rate_bps) != -1) { t->result = 2; break; }
         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;
@@ -534,6 +551,7 @@ static void call_tick(void* arg) {
         // Завершение во время незавершённой сборки/паузы приёма не оставляет callbacks.
         g_video_ready = 0;
         {
+            if (call_video_configure_rate(A, t->call_id, 3, 400000, 400000)) { t->result = 2; break; }
             struct call_video_frame f = video_test_frame(t->call_id, 3);
             if (call_video_send_frame(A, &f) != 1) { t->result = 2; break; }
         }