Retain video mailbox tails and pause raw input under backpressure

src/call/call_video_port.c

@@ -26,6 +26,7 @@ static struct {
     uint32_t minimum, maximum;
     struct call_video_frame tx[CALL_VIDEO_PORT_TX_FRAMES], rx[8];
     unsigned tx_count;
+    int tx_processing;
     unsigned count, bytes;
     uint64_t decoded_generation;
     int decoded_pending, decoded_accepted;
@@ -86,7 +87,15 @@ static void decoded(void *arg, uint64_t generation, int accepted) {
     }
     pthread_mutex_unlock(&lock);
 }
-static const struct call_video_ops ops = {peer, ready, frame, need_key, bitrate, decoded};
+static void writable(void *arg, int admit) {
+    pthread_mutex_lock(&lock);
+    if (port.id == (uint64_t)(uintptr_t)arg) {
+        port.status.can_encode = admit;
+        if (port.tx_count && !port.tx_processing) schedule();
+    }
+    pthread_mutex_unlock(&lock);
+}
+static const struct call_video_ops ops = {peer, ready, frame, need_key, bitrate, decoded, writable};
 
 /* Снимок команды под mutex; API транспорта вызывается без mutex, callbacks могут войти обратно. */
 static void process(void *arg) {
@@ -106,7 +115,7 @@ static void process(void *arg) {
     uint64_t generation = port.applied_generation, request = port.applied_request;
     struct call_video_frame tx[CALL_VIDEO_PORT_TX_FRAMES];
     unsigned tx_count = port.tx_count;
-    memcpy(tx, port.tx, tx_count * sizeof(*tx)); port.tx_count = 0;
+    memcpy(tx, port.tx, tx_count * sizeof(*tx)); port.tx_count = 0; port.tx_processing = tx_count != 0;
     port.opening = port.configure = port.completed = port.resume = port.key = 0;
     port.range = 0;
     port.decoded_pending = 0;
@@ -147,19 +156,25 @@ static void process(void *arg) {
             sent = call_video_send_frame(inst, &tx[sent_count]);
             if (sent != 1) break;
         }
-        if (sent_count < tx_count)
-            DEBUG_WARN(DEBUG_CATEGORY_CALL, "video port TX frames released call=%llu accepted=%u unsent=%u result=%d "
+        if (sent_count < tx_count && sent < 0)
+            DEBUG_WARN(DEBUG_CATEGORY_CALL, "video port TX chain failed call=%llu accepted=%u unsent=%u result=%d "
                 "first_unsent=%llu last=%llu", (unsigned long long)id, sent_count, tx_count - sent_count, sent,
                 (unsigned long long)tx[sent_count].frame_id, (unsigned long long)tx[tx_count - 1].frame_id);
     }
-    for (unsigned i = 0; i < tx_count; ++i) u_free((void *)tx[i].data);
     pthread_mutex_lock(&lock);
+    port.tx_processing = 0;
     if (port.id == id) {
         port.status.minimum_bps = minimum; port.status.maximum_bps = maximum;
         if (result < 0) { port.status.error = 1; DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port configuration failed call=%llu", (unsigned long long)id); }
         else if (configure) { port.status.configured_generation = format.generation; port.status.error = 0; }
-        if (sent != 1) ++port.status.key_serial;
+        if (!sent && active) {
+            port.tx_count = tx_count - sent_count;
+            memcpy(port.tx, tx + sent_count, port.tx_count * sizeof(*tx));
+        }
+        if (sent < 0) ++port.status.key_serial;
     }
+    unsigned release = !sent && active && port.id == id ? sent_count : tx_count;
+    for (unsigned i = 0; i < release; ++i) u_free((void *)tx[i].data);
     pthread_mutex_unlock(&lock);
 }
 int call_video_port_attach(struct UTUN_INSTANCE *inst) {
@@ -205,7 +220,7 @@ void call_video_port_close(uint64_t id) {
 int call_video_port_status(uint64_t id, struct call_video_port_status *out) {
     pthread_mutex_lock(&lock);
     int result = id && id == port.id && port.inst && out ? 0 : -1;
-    if (!result) *out = port.status;
+    if (!result) { *out = port.status; out->can_encode = out->can_encode && !port.tx_count && !port.tx_processing; }
     pthread_mutex_unlock(&lock); return result;
 }
 int call_video_port_format(uint64_t id, const struct call_video_format *format, uint32_t bps) {
@@ -237,7 +252,7 @@ int call_video_port_send(const struct call_video_frame *frames, unsigned count)
     }
     pthread_mutex_lock(&lock);
     int result = frames[0].call_id && frames[0].call_id == port.id && port.inst ? 0 : -1;
-    if (!result && !port.tx_count) {
+    if (!result && !port.tx_count && !port.tx_processing) {
         unsigned copied = 0;
         for (; copied < count; ++copied) {
             uint8_t *data = u_malloc(frames[copied].size);
@@ -253,9 +268,9 @@ int call_video_port_send(const struct call_video_frame *frames, unsigned count)
             result = -1;
         }
     }
-    if (result == 0) DEBUG_WARN(DEBUG_CATEGORY_CALL,
-        "video port TX mailbox busy call=%llu frames=%u first=%llu pending=%u; batch not queued",
-        (unsigned long long)frames[0].call_id, count, (unsigned long long)frames[0].frame_id, port.tx_count);
+    if (result == 0) DEBUG_TRACE(DEBUG_CATEGORY_CALL,
+        "video port TX retry call=%llu frames=%u first=%llu pending=%u processing=%d",
+        (unsigned long long)frames[0].call_id, count, (unsigned long long)frames[0].frame_id, port.tx_count, port.tx_processing);
     pthread_mutex_unlock(&lock); return result;
 }
 int call_video_port_pull(uint64_t id, struct call_video_frame *out) {

src/call/call_video_port.h

@@ -16,6 +16,7 @@ struct call_video_port_status {
     uint32_t rate_bps;
     uint32_t minimum_bps, maximum_bps;
     int have_peer, error;
+    int can_encode; // Новые raw входы допустимы; готовый TX всё равно сохраняется и повторяется.
     uint64_t decoded_generation;
     int decoded_accepted;
 };

tools/chatgui/src/call_video_engine.cpp

@@ -191,6 +191,10 @@ void CallVideoEngine::transmit() {
     int64_t reportAt = nowUs(), encodeMax = 0, inputAgeMax = 0;
     uint64_t captureBase = 0;
     unsigned encodedCount = 0, keyCount = 0, sendBusy = 0, expired = 0;
+    QList<LiveEncodedFrame> pendingOutputs;
+    call_video_frame pendingPackets[CALL_VIDEO_PORT_TX_FRAMES]{};
+    unsigned pendingCount = 0;
+    uint64_t pendingRevision = 0;
     uint64_t wakeSerial = 0;
     auto idle = [this, &wakeSerial] {
         std::unique_lock<std::mutex> guard(m->lock);
@@ -208,6 +212,18 @@ void CallVideoEngine::transmit() {
           clockCorrection = m->clockCorrection; image = m->capturedImage; wakeSerial = m->wakeSerial;
         }
         int64_t now = nowUs();
+        if (pendingCount) {
+            if (!enabled || revision != pendingRevision) {
+                DEBUG_INFO(DEBUG_CATEGORY_CALL, "video TX cancelled by mode change call=%llu frames=%u",
+                    (unsigned long long)m->id, pendingCount);
+                pendingCount = 0; pendingOutputs.clear(); forceKey = true;
+            } else {
+                int sent = call_video_port_send(pendingPackets, pendingCount);
+                if (!sent) { ++sendBusy; idle(); continue; }
+                if (sent < 0) { forceKey = true; m->report(tr("Video transport rejected encoded frames")); }
+                pendingCount = 0; pendingOutputs.clear();
+            }
+        }
         if (now - reportAt >= 1000000) {
             DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
                 "video capture flow call=%llu interval_us=%lld captured=%llu encoded=%u keys=%u send_busy=%u "
@@ -258,6 +274,7 @@ void CallVideoEngine::transmit() {
                 call_video_port_applied(m->id, format.generation, request, 0);
         }
 #ifdef HAVE_STREAM_CAPTURE
+        if (!state.can_encode) { idle(); continue; }
         if (image.isNull() || nowUs() - pts > CallCameraMaxFrameAgeUs) {
             DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX skips old capture age_us=%lld", (long long)(nowUs() - pts));
             lastPts = pts; ++expired; continue;
@@ -279,8 +296,9 @@ void CallVideoEngine::transmit() {
             DEBUG_ERROR(DEBUG_CATEGORY_CALL, "GPU output batch exceeds mailbox frames=%lld", (long long)outputs.size());
             forceKey = true; continue;
         }
-        call_video_frame packets[CALL_VIDEO_PORT_TX_FRAMES]{}; unsigned count = 0;
-        for (const auto& output : outputs) {
+        pendingOutputs = std::move(outputs); pendingRevision = revision;
+        auto* packets = pendingPackets; unsigned count = 0;
+        for (const auto& output : pendingOutputs) {
             ++encodedCount; if (output.key) ++keyCount;
             auto& packet = packets[count++];
             packet.call_id = m->id; packet.frame_id = ++frameId; packet.pts_us = output.pts; packet.format = format;
@@ -288,15 +306,15 @@ void CallVideoEngine::transmit() {
             packet.flags = output.key ? CALL_VIDEO_KEY : 0;
         }
         { std::lock_guard<std::mutex> guard(m->lock);
-          if (!m->running || !m->enabled || m->revision != revision || nowUs() - packets[0].pts_us > CallCameraMaxFrameAgeUs) {
+          if (!m->running || !m->enabled || m->revision != revision) {
               DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX discards obsolete batch revision=%llu/%llu enabled=%d age_us=%lld frames=%u",
                   (unsigned long long)revision, (unsigned long long)m->revision, m->enabled,
                   (long long)(nowUs() - packets[0].pts_us), count);
-              expired += count; forceKey = true; continue;
+              expired += count; forceKey = true; pendingOutputs.clear(); continue;
           }
-          forceKey = call_video_port_send(packets, count) != 1; }
+          pendingCount = count; forceKey = false; }
         for (unsigned i = 0; i < count; ++i) {
-            const auto& output = outputs[i];
+            const auto& output = pendingOutputs[i];
             // Одинаковое окно исходного PTS на всех этапах: примерно один кадр в секунду, без трассы каждого кадра.
             if (output.pts % 1000000 < 34000)
                 DEBUG_DEBUG(DEBUG_CATEGORY_CALL,