From 56a06da35f96ffe92358cc61ee77b09b13db84af Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 13:46:33 +0100 Subject: [PATCH] Cover call admission, ID reuse and retained sessions on live cores --- tests/Makefile.am | 6 + tests/test_call_admission.c | 302 ++++++++++++++++++++++++++++++++++++ 2 files changed, 308 insertions(+) create mode 100644 tests/test_call_admission.c diff --git a/tests/Makefile.am b/tests/Makefile.am index 0da2a3db..a91dd361 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -134,6 +134,7 @@ check_PROGRAMS += \ test_stream_service \ test_stream_member \ test_stream_publisher \ + test_call_admission \ test_u_async_comprehensive \ test_nat_detection \ test_audio_compressor \ @@ -933,6 +934,11 @@ test_call_video_flush_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) test_call_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/src/dm -I$(top_srcdir)/src/call -I$(top_srcdir)/src/routing_layer -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/lib test_call_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +test_call_admission_SOURCES = test_call_admission.c $(test_clock_sources) +test_call_admission_CFLAGS = $(test_call_CFLAGS) +test_call_admission_LDADD = $(test_call_LDADD) +test_call_admission_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=etcp_route_send -Wl,--wrap=conn_mgr_open + test_call_headless_SOURCES = test_call_headless.c test_call_headless_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/src/dm -I$(top_srcdir)/src/call -I$(top_srcdir)/src/routing_layer -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/lib test_call_headless_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) diff --git a/tests/test_call_admission.c b/tests/test_call_admission.c new file mode 100644 index 00000000..e8ebb1e5 --- /dev/null +++ b/tests/test_call_admission.c @@ -0,0 +1,302 @@ +/* Настоящие CHAT/BGP/ETCP/CM из существующего fixture; отдельный процесс и собственная state machine. */ +#define main call_reference_main +#include "test_call.c" +#undef main +#include "test_clock.h" +#include "../src/radio/radio_call.h" + +#define ADMISSION_FIRST UINT64_C(0x200000001) +#define ADMISSION_SECOND UINT64_C(0x200000002) +#define ADMISSION_THIRD UINT64_C(0x200000003) +#define ADMISSION_RADIO UINT64_C(0x300000001) + +struct admission_events { + unsigned count, ended, errors, invites, cm_open, radio_finished; + uint64_t ended_id; + uint8_t ended_reason, radio_reason; +}; +struct admission_test { + struct UASYNC *ua; + struct UTUN_INSTANCE *inst[2]; + struct admission_events events[2]; + struct ETCP_CONN *transport; + void *timer; + char dm[2][64]; + unsigned phase, ticks, checks, refusals, duplicates, dropped_acks; + uint64_t next_id; + int hold_acks, result; +}; +static struct admission_test *admission_current; + +#define ADMISSION_CHECK(t, expression) do { \ + (t)->checks++; \ + if (!(expression)) { \ + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call admission assertion line=%d phase=%u: %s", __LINE__, (t)->phase, #expression); \ + return 0; \ + } \ +} while (0) + +static int admission_index(struct UTUN_INSTANCE *inst) { + if (!admission_current) return -1; + for (int i = 0; i < 2; i++) if (admission_current->inst[i] == inst) return i; + return -1; +} + +int __real_etcp_route_send(struct UTUN_INSTANCE *, uint64_t, uint64_t, struct ll_entry *, int, int); +int __wrap_etcp_route_send(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, struct ll_entry *e, int force, int mode) { + int index = admission_index(inst); + if (index >= 0 && e && e->dgram && e->len >= 2 && e->dgram[0] == ETCP_RT_ID_CALL) { + if (e->dgram[1] == CALL_SUBCMD_INVITE) admission_current->events[index].invites++; + if (e->dgram[1] == CALL_SUBCMD_END_ACK && admission_current->hold_acks) { + admission_current->dropped_acks++; + queue_dgram_free(e); queue_entry_free(e); return 0; + } + } + return __real_etcp_route_send(inst, group, peer, e, force, mode); +} + +int __real_conn_mgr_open(struct UTUN_INSTANCE *, uint64_t, uint64_t, conn_mgr_cb_t, void *, struct CONN_MGR_HANDLE **); +int __wrap_conn_mgr_open(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, conn_mgr_cb_t cb, void *arg, + struct CONN_MGR_HANDLE **out) { + int index = admission_index(inst); + if (index >= 0) admission_current->events[index].cm_open++; + return __real_conn_mgr_open(inst, group, peer, cb, arg, out); +} + +static void admission_event(struct UTUN_INSTANCE *inst, int type, const uint8_t *data, int length) { + int index = admission_index(inst); + if (index < 0) return; + struct admission_events *ev = &admission_current->events[index]; + switch (type) { + case CHAT_EVT_CALL_ENDED: + if (length >= 9) { memcpy(&ev->ended_id, data, 8); ev->ended_reason = data[8]; ev->ended++; } + /* fall through */ + case CHAT_EVT_CALL_INCOMING: case CHAT_EVT_CALL_RINGING: case CHAT_EVT_CALL_ACCEPTED: + case CHAT_EVT_CALL_DECLINED: case CHAT_EVT_CALL_PHONE_BUSY: case CHAT_EVT_CALL_PATH: + case CHAT_EVT_CALL_CONNECTION: case CHAT_EVT_CALL_STATS: + ev->count++; break; + case CHAT_EVT_CALL_ERROR: ev->count++; ev->errors++; break; + case CHAT_EVT_RADIO_CALL: + if (length == 26 && data[24] == RADIO_CALL_FINISHED) { ev->radio_finished++; ev->radio_reason = data[25]; } + break; + default: break; + } + call_evt_handler(inst, type, data, length); +} + +static unsigned admission_live(struct UTUN_INSTANCE *inst) { + unsigned count = 0; + for (struct ll_entry *e = inst->call->sessions->head; e; e = e->next) + count += ((struct call_session *)e->data)->state != CALL_ST_ENDED; + return count; +} + +/* Отказ не меняет существующий звонок, CM, таймеры, очередь и счётчики. */ +static int admission_attempt(struct admission_test *t, int index, uint64_t original, const char *ch, uint64_t peer, + uint64_t requested, int busy_event) { + struct UTUN_INSTANCE *inst = t->inst[index]; + struct call_session *s = call_session_find(inst->call, original); + struct call_session saved = {0}; + if (s) saved = *s; + struct admission_events before = t->events[index]; + uint32_t started = inst->call->c_calls_started; + int records = queue_entry_count(inst->call->sessions); + unsigned live = admission_live(inst); + call_start(inst, ch, peer, requested); + struct admission_events *after = &t->events[index]; + ADMISSION_CHECK(t, after->cm_open == before.cm_open && after->invites == before.invites); + ADMISSION_CHECK(t, inst->call->c_calls_started == started && queue_entry_count(inst->call->sessions) == records); + ADMISSION_CHECK(t, admission_live(inst) == live && after->errors == before.errors); + if (s) { + ADMISSION_CHECK(t, call_session_find(inst->call, original) == s); + ADMISSION_CHECK(t, s->state == saved.state && s->role == saved.role && s->reason == saved.reason); + ADMISSION_CHECK(t, s->group_id == saved.group_id && s->peer_node_id == saved.peer_node_id && s->force == saved.force); + ADMISSION_CHECK(t, s->cm_handle == saved.cm_handle && s->tx_q == saved.tx_q && s->started_tb == saved.started_tb); + ADMISSION_CHECK(t, s->ring_timer == saved.ring_timer && s->invite_timer == saved.invite_timer); + ADMISSION_CHECK(t, s->traffic_timer == saved.traffic_timer && s->teardown_timer == saved.teardown_timer); + ADMISSION_CHECK(t, s->stats_timer == saved.stats_timer && s->tx_media_seq == saved.tx_media_seq); + } + if (busy_event) { + ADMISSION_CHECK(t, !call_session_find(inst->call, requested)); + ADMISSION_CHECK(t, after->count == before.count + 1 && after->ended == before.ended + 1); + ADMISSION_CHECK(t, after->ended_id == requested && after->ended_reason == CALL_REASON_BUSY); + t->refusals++; + } else { + ADMISSION_CHECK(t, after->count == before.count && after->ended == before.ended); + t->duplicates++; + } + return 1; +} + +static int admission_busy_cases(struct admission_test *t, int index, uint64_t original) { + struct UTUN_INSTANCE *inst = t->inst[index]; + uint64_t peer = t->inst[1 - index]->node_id; + ADMISSION_CHECK(t, call_busy(inst) && admission_live(inst) == 1); + ADMISSION_CHECK(t, admission_attempt(t, index, original, CH_ID, peer, original, 0)); + /* Даже другой tuple с тем же ID не должен завершать/дублировать текущую запись. */ + ADMISSION_CHECK(t, admission_attempt(t, index, original, "4242424242424243", peer ^ 1, original, 0)); + ADMISSION_CHECK(t, admission_attempt(t, index, original, CH_ID, peer, ++t->next_id, 1)); + ADMISSION_CHECK(t, admission_attempt(t, index, original, t->dm[index], peer, ++t->next_id, 1)); + /* Занятость instance проверяется до поиска другой группы/доступности другого пира. */ + ADMISSION_CHECK(t, admission_attempt(t, index, original, "4242424242424243", peer ^ 1, ++t->next_id, 1)); + DEBUG_INFO(DEBUG_CATEGORY_CALL, "call admission protected node=%d id=%llx state=%u role=%u live=1 no_new_CM_INVITE=1", + index, (unsigned long long)original, call_session_find(inst->call, original)->state, + call_session_find(inst->call, original)->role); + return 1; +} + +/* 1 — продолжить, 2 — завершить; 0 — ошибка. Состояния меняются только штатным сигналингом. */ +static int admission_step(struct admission_test *t) { + struct UTUN_INSTANCE *a = t->inst[0], *b = t->inst[1]; + struct call_session *sa = call_session_find(a->call, t->phase < 6 ? ADMISSION_FIRST : ADMISSION_SECOND); + struct call_session *sb = call_session_find(b->call, t->phase < 6 ? ADMISSION_FIRST : ADMISSION_SECOND); + switch (t->phase) { + case 0: + if (!conn_up(a, b->node_id) || !conn_up(b, a->node_id)) return 1; + do_setup(a, IDX_A); do_setup(b, IDX_B); t->phase++; break; + case 1: + if (!bgp_has(a, b->node_id) || !bgp_has(b, a->node_id)) return 1; + for (int i = 0; i < 2; i++) { + uint64_t peer = t->inst[1 - i]->node_id; + ADMISSION_CHECK(t, dm_start(t->inst[i], peer, g_sh.x_pub[1 - i], g_sh.ed_pub[1 - i], "peer", CH_ID) == 0); + snprintf(t->dm[i], sizeof(t->dm[i]), "dm:%llu", (unsigned long long)dm_derive_conv_id(t->inst[i]->node_id, peer)); + } + t->transport = instance_find_conn(a, b->node_id); + ADMISSION_CHECK(t, admission_attempt(t, 0, 0, CH_ID, b->node_id, 0, 0)); + ADMISSION_CHECK(t, admission_attempt(t, 0, 0, CH_ID, 0, ++t->next_id, 0)); + call_start(a, CH_ID, b->node_id, ADMISSION_FIRST); + sa = call_session_find(a->call, ADMISSION_FIRST); + ADMISSION_CHECK(t, sa && sa->state == CALL_ST_OUTGOING); + ADMISSION_CHECK(t, admission_busy_cases(t, 0, ADMISSION_FIRST)); + t->phase++; break; + case 2: + if (!sa || !sb || sa->state != CALL_ST_RINGING || sb->state != CALL_ST_RINGING) return 1; + ADMISSION_CHECK(t, admission_busy_cases(t, 0, ADMISSION_FIRST)); + ADMISSION_CHECK(t, admission_busy_cases(t, 1, ADMISSION_FIRST)); + call_accept(b, ADMISSION_FIRST); t->phase++; break; + case 3: + if (!sa || !sb || sa->state != CALL_ST_ACTIVE || sb->state != CALL_ST_ACTIVE) return 1; + ADMISSION_CHECK(t, sa->cm_handle && sb->cm_handle); + ADMISSION_CHECK(t, admission_busy_cases(t, 0, ADMISSION_FIRST)); + ADMISSION_CHECK(t, admission_busy_cases(t, 1, ADMISSION_FIRST)); + { + unsigned before = t->events[0].radio_finished; + struct radio_call_request req = {a, ADMISSION_RADIO, sa->group_id, b->node_id, RADIO_CALL_START}; + radio_call_execute(&req); + ADMISSION_CHECK(t, !radio_call_busy(a) && call_busy(a)); + ADMISSION_CHECK(t, t->events[0].radio_finished == before + 1 && t->events[0].radio_reason == RADIO_CALL_BUSY); + } + { + const uint8_t opus[] = {0xf8, 0xff, 0xfe}; + ADMISSION_CHECK(t, call_send_media(a, ADMISSION_FIRST, opus, sizeof(opus), &sent_timing) == 0); + ADMISSION_CHECK(t, call_send_media(b, ADMISSION_FIRST, opus, sizeof(opus), &sent_timing) == 0); + } + t->phase++; break; + case 4: + if (!g_media[0].frames || !g_media[1].frames) return 1; + ADMISSION_CHECK(t, g_media[0].last_call_id == ADMISSION_FIRST && g_media[1].last_call_id == ADMISSION_FIRST); + ADMISSION_CHECK(t, g_media[0].last_len == 3 && !memcmp(g_media[0].last, g_media[1].last, 3)); + ADMISSION_CHECK(t, instance_find_conn(a, b->node_id) == t->transport && conn_up(a, b->node_id)); + t->hold_acks = 1; call_hangup(a, ADMISSION_FIRST); t->phase++; break; + case 5: + if (!sa || !sb || sa->state != CALL_ST_ENDED || sb->state != CALL_ST_ENDED) return 1; + ADMISSION_CHECK(t, t->dropped_acks && !call_busy(a) && !call_busy(b)); + ADMISSION_CHECK(t, admission_attempt(t, 0, ADMISSION_FIRST, CH_ID, b->node_id, ADMISSION_FIRST, 0)); + ADMISSION_CHECK(t, admission_attempt(t, 1, ADMISSION_FIRST, t->dm[1], a->node_id, ADMISSION_FIRST, 0)); + call_set_phone_state(b, 2, CALL_POLICY_WARN_PHONE); + call_start(a, t->dm[0], b->node_id, ADMISSION_SECOND); + ADMISSION_CHECK(t, call_session_find(a->call, ADMISSION_FIRST) == sa && sa->state == CALL_ST_ENDED); + sa = call_session_find(a->call, ADMISSION_SECOND); + ADMISSION_CHECK(t, sa && sa->state == CALL_ST_OUTGOING && admission_live(a) == 1); + t->phase++; break; + case 6: + if (!sa || !sb || sa->state != CALL_ST_WAIT_DECISION || sb->state != CALL_ST_WAIT_DECISION) return 1; + ADMISSION_CHECK(t, admission_busy_cases(t, 0, ADMISSION_SECOND)); + ADMISSION_CHECK(t, admission_busy_cases(t, 1, ADMISSION_SECOND)); + call_continue(a, ADMISSION_SECOND); t->phase++; break; + case 7: + if (!sa || !sb || sa->state != CALL_ST_RINGING || sb->state != CALL_ST_RINGING) return 1; + ADMISSION_CHECK(t, sa->force && sb->force); + call_hangup(a, ADMISSION_SECOND); t->hold_acks = 0; t->phase++; break; + case 8: + if (queue_entry_count(a->call->sessions) || queue_entry_count(b->call->sessions)) return 1; + call_set_phone_state(b, 0, CALL_POLICY_PUBLISH_PHONE); + { + struct radio_call_request req = {a, ADMISSION_RADIO + 1, strtoull(CH_ID, NULL, 10), b->node_id, RADIO_CALL_START}; + radio_call_execute(&req); + ADMISSION_CHECK(t, radio_call_busy(a) && !call_busy(a)); + ADMISSION_CHECK(t, admission_attempt(t, 0, 0, CH_ID, b->node_id, ++t->next_id, 1)); + req.action = RADIO_CALL_CANCEL; radio_call_execute(&req); + } + t->phase++; break; + case 9: + if (radio_call_busy(a) || radio_call_busy(b)) return 1; + call_start(a, CH_ID, b->node_id, ADMISSION_THIRD); t->phase++; break; + case 10: + sa = call_session_find(a->call, ADMISSION_THIRD); sb = call_session_find(b->call, ADMISSION_THIRD); + if (!sa || !sb || sa->state != CALL_ST_RINGING || sb->state != CALL_ST_RINGING) return 1; + call_decline(b, ADMISSION_THIRD); t->phase++; break; + case 11: + if (queue_entry_count(a->call->sessions) || queue_entry_count(b->call->sessions)) return 1; + ADMISSION_CHECK(t, !call_busy(a) && !call_busy(b)); + ADMISSION_CHECK(t, instance_find_conn(a, b->node_id) == t->transport && conn_up(a, b->node_id)); + ADMISSION_CHECK(t, t->refusals == 22 && t->duplicates == 18); + DEBUG_INFO(DEBUG_CATEGORY_CALL, "call admission PASS checks=%u BUSY=%u no_event=%u original_audio_RX=%d/%d END_ACK_drops=%u", + t->checks, t->refusals, t->duplicates, g_media[0].frames, g_media[1].frames, t->dropped_acks); + return 2; + } + return 1; +} + +static void admission_tick(void *arg) { + struct admission_test *t = arg; t->timer = NULL; + int result = admission_step(t); + if (++t->ticks > 3000) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call admission timeout phase=%u", t->phase); result = 0; + } + if (result == 1) { + t->timer = uasync_set_timeout(t->ua, TICK_TB, t, admission_tick, "call_admission"); + if (t->timer) return; + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call admission timer allocation failed"); result = 0; + } + t->result = result == 2 ? 0 : 1; uasync_stop(t->ua); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); + debug_set_category_level(DEBUG_CATEGORY_CALL, DEBUG_LEVEL_INFO); + utun_instance_set_tun_init_enabled(0); test_clock_init(TICK_TB); test_clock_sync_wall_time(); + size_t allocated = u_get_allocated_count(); + char dir[512] = "/tmp/utun_call_admission_XXXXXX"; + if (test_mkdtemp(dir) || test_setenv("UTUN_TEST_DIR", dir, 1)) return 1; + srand(1); fill_shared(); write_configs(dir); + struct admission_test t = {.next_id = UINT64_C(0x400000000), .result = 1}; + admission_current = &t; + t.ua = uasync_create(); + for (int i = 0; t.ua && i < 2; i++) { + char path[600]; snprintf(path, sizeof(path), "%s/%s.conf", dir, i ? "b" : "a"); + t.inst[i] = utun_instance_create(t.ua, path); + } + if (t.inst[0] && t.inst[1]) { + test_etcp_set_peer(t.inst[1], "to_a", t.inst[0], "s1"); + for (int i = 0; i < 2; i++) { + utun_instance_init(t.inst[i]); t.inst[i]->test_user_ptr = &g_ev[i]; + chat_event_set_handler(t.inst[i], admission_event); + call_set_media_recv_cb(t.inst[i], media_recv_cb, &g_media[i]); + } + t.timer = uasync_set_timeout(t.ua, TICK_TB, &t, admission_tick, "call_admission"); + if (t.timer) test_clock_mainloop(t.ua); + } + if (t.timer) uasync_cancel_timeout(t.ua, t.timer); + for (int i = 0; i < 2; i++) if (t.inst[i]) utun_instance_destroy(t.inst[i]); + if (t.ua) { uasync_poll(t.ua, 0); uasync_destroy(t.ua, 0); } + admission_current = NULL; test_clock_finish(); + char command[600]; snprintf(command, sizeof(command), "rm -rf %s", dir); (void)!system(command); + size_t remaining = u_get_allocated_count(); + if (remaining != allocated) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call admission leaked allocations before=%zu after=%zu", allocated, remaining); t.result = 1; + } + printf("call admission: %s, checks=%u BUSY=%u no_event=%u allocations=%zu->%zu\n", + t.result ? "FAIL" : "PASS", t.checks, t.refusals, t.duplicates, allocated, remaining); + return t.result; +}