Expose pending source deliveries without counting ETCP inflight twice

src/stream/stream_media_buffer.c

@@ -159,7 +159,7 @@ int stream_media_store_feed(struct stream_media_store *s, uint64_t seq, const ui
     int result = stream_media_parse(data, length, &b);
     if (result != STREAM_OK) { s->failed = 1; return result; }
     if (s->configured && (b.format.audio_codec != s->format.audio_codec || b.format.revision < s->format.revision
-        || (b.format.revision == s->format.revision && (b.format.width != s->format.width
+        || (b.format.revision == s->format.revision && (b.format.video_codec != s->format.video_codec || b.format.width != s->format.width
             || b.format.height != s->format.height || b.format.rotation != s->format.rotation)))) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index conflicting format source=%llx revision=%u/%u audio=%u/%u",
                     (unsigned long long)s->descriptor.source_node_id, s->format.revision, b.format.revision,

src/stream/stream_network.c

@@ -1806,6 +1806,15 @@ static void pressure_tick(struct stream_network *net, uint64_t now) {
     }
 }
 
+uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d) {
+    struct stream_network *net = stream_network_get(inst); uint64_t bytes = 0;
+    if (!net || !d) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "pending stream snapshot unavailable"); return 0; }
+    for (struct stream_peer *p = net->peers; p; p = p->next)
+        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
+            if (!f->outgoing && same_stream(&f->descriptor, d)) bytes += f->load.queued.bytes;
+    return bytes;
+}
+
 int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t node, struct stream_peer_load_info *info) {
     struct stream_network *net = stream_network_get(inst); struct stream_peer *p = net ? peer_find(net, node) : NULL;
     if (!p || !info) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream load unavailable peer=%llx", (unsigned long long)node); return STREAM_ERR_STATE; }
@@ -1819,10 +1828,11 @@ int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t node, struct strea
     result.load.removed = p->load.removed; result.load.rejected = p->load.rejected;
     struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd);
     if (conn) {
-        struct ll_queue *queues[] = {conn->send_input_q, conn->input_queue, conn->input_send_q};
+        struct ll_queue *queues[] = {conn->send_input_q, conn->input_queue};
-        for (unsigned i = 0; i < 3; i++) if (queues[i]) {
+        for (unsigned i = 0; i < 2; i++) if (queues[i]) {
             result.transport_queued_packets += queues[i]->count; result.transport_queued_bytes += queues[i]->total_bytes;
         }
+        result.transport_queued_bytes = etcp_tx_queued(conn);
         result.inflight_bytes = conn->unacked_bytes;
     }
     *info = result; return STREAM_OK;

src/stream/stream_network.h

@@ -55,6 +55,8 @@ struct stream_peer_load_info {
     int global_pressure, request_active, accepting;
 };
 int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t peer, struct stream_peer_load_info *info);
+/* Оставшиеся RS-части всех локальных выдач точной публикации; принятые ETCP части сюда не входят. */
+uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor);
 
 int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out);
 void stream_network_destroy(struct stream_network *network);