Record aggregate Router and transport histories and verify weighted audio statistics

src/call/call.c

@@ -28,6 +28,8 @@
 #include "../routing_layer/topo_node.h"
 #include "../transport_layer/etcp_api.h"
 #include "../transport_layer/etcp.h"
+#include "../transport_layer/etcp_tx_stats.h"
+#include "call_audio_history.h"
 #include "../../lib/mem.h"
 #include "../../lib/ll_queue.h"
 #include "../../lib/debug_config.h"
@@ -1012,6 +1014,21 @@ static void call_audio_batch_tick(void* arg) {
     struct ETCP_ROUTER_TX_STATS net = {0};
     uint64_t now = get_time_us();
     int found = etcp_router_tx_stats(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_AUDIO, now, &net) == 0;
+    const struct call_audio_ops* audio = s->ctx->inst->call_audio_ops;
+    if (audio && audio->record_tx) {
+        struct call_history_tx snapshot = {.at_ms = now / 1000, .queued_bytes = s->tx_q ? s->tx_q->total_bytes : 0};
+        struct ll_queue* queues = s->ctx->inst->router_conns;
+        for (struct ll_entry* e = queues ? queues->head : NULL; e; e = e->next) {
+            struct ETCP_ROUTER_CONN* r = (struct ETCP_ROUTER_CONN*)e;
+            if (r->closed) continue;
+            if (r->send_q) snapshot.queued_bytes += r->send_q->total_bytes;
+            if (r->inflight_q) snapshot.inflight_bytes += r->inflight_q->total_bytes;
+        }
+        struct etcp_tx_snapshot transport;
+        if (etcp_tx_get_snapshot(s->ctx->inst, now, &transport) == 0)
+            snapshot.transport_bytes = transport.queued_bytes + transport.inflight_bytes;
+        audio->record_tx(s->call_id, &snapshot);
+    }
     if (call_audio_batch_step(&s->audio_batch, now, found ? &net : NULL))
         DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio batch control failed call=%llx; retaining target=%u",
                     (unsigned long long)s->call_id, s->audio_batch.target);

tools/chatgui/tests/test_call_jitter.cpp

@@ -2,6 +2,7 @@
 #include "call/call_jitter.h"
 #include "call/call_jitter_window.h"
 #include "call/call_playout_policy.h"
+#include "call/call_history_store.h"
 #include "call/call_audio.h"
 #include "utun_instance.h"
 #include "../../../lib/opus_codec.h"
@@ -195,7 +196,33 @@ static void test_render_readiness() {
     call_audio_release(id); opus_codec_encoder_destroy(enc);
 }
 
+static void test_audio_history() {
+    CallHistoryStore history;
+    call_audio_history snapshot;
+    history.receive(0, 100, 200, 100, 0, 0); // Подготовка не получает веса тишины.
+    history.receive(10, 100, 200, 100, 480, 240);
+    history.receive(20, 200, 200, 100, 480, 0);
+    history.receive(10020, 300, 200, 100, 960, 0);
+    history.transmit({10020, 100, 300, 250});
+    history.snapshot(10020, snapshot);
+    CHECK(snapshot.rx_count == 2 && snapshot.tx_count == 1);
+    CHECK(snapshot.rx[0].low_ms == 100 && snapshot.rx[0].high_ms == 200);
+    CHECK(std::abs(snapshot.observed_ms - 30) < .001);
+    CHECK(std::abs(snapshot.missing_ms - 2.5) < .001);
+    CHECK(std::abs(snapshot.histogram[2] - 1.0 / 6) < .001);
+    CHECK(std::abs(snapshot.histogram[6] - 2.0 / 3) < .001);
+    CHECK(snapshot.tx[0].queued_bytes + snapshot.tx[0].inflight_bytes == 400); // Транспорт не складывается повторно.
+    history.snapshot(40020, snapshot);
+    CHECK(!snapshot.rx_count && !snapshot.tx_count && snapshot.observed_ms == 0);
+    for (int t = 50000; t < 90000; t += 50) history.receive(t, 4000, 200, 100, 2400, 0);
+    history.snapshot(89950, snapshot);
+    CHECK(snapshot.rx_count == 600 && snapshot.rx[0].at_ms == 60000);
+    CHECK(std::abs(snapshot.histogram[60] - 1) < .001);
+    puts("audio history: time weights, startup exclusion, gaps, overflow bin and ring expiry PASS");
+}
+
 int main() {
+    test_audio_history();
     test_startup_policy();
     test_packet_arrival_sampling();
     CHECK(call_audio_init(&audio_owner) == 0);