Cancel pending radio batches when CHAT membership is revoked

src/radio/radio.c

@@ -280,6 +280,27 @@ static void radio_talker_off(struct radio_ctx* ctx, uint64_t src) {
     }
 }
 
+void radio_revalidate_members(struct TOPO_GROUP* group) {
+    struct radio_ctx* ctx = radio_of(group);
+    if (!ctx || group->group_type != TOPO_GROUP_TYPE_CHAT) return;
+    int self_allowed = topo_group_chat_allowed(group->instance, group->group_id, group->instance->node_id);
+    if (!self_allowed) {
+        if (ctx->talking || ctx->burst_count)
+            DEBUG_INFO(DEBUG_CATEGORY_RADIO, "radio: revoke local transmission group=%llx stream=%u buffered=%u",
+                       (unsigned long long)group->group_id, ctx->stream_id, ctx->burst_count);
+        if (ctx->burst_timer) uasync_cancel_timeout(group->instance->ua, ctx->burst_timer);
+        ctx->burst_timer = NULL; ctx->burst_count = 0; ctx->burst_bytes = 0; ctx->talking = 0;
+    }
+    for (unsigned i = 0; i < RADIO_MAX_TALKERS; i++) {
+        uint64_t source = ctx->talkers[i].src;
+        if (source && (!self_allowed || !topo_group_chat_allowed(group->instance, group->group_id, source))) {
+            DEBUG_INFO(DEBUG_CATEGORY_RADIO, "radio: revoke talker group=%llx source=%llx",
+                       (unsigned long long)group->group_id, (unsigned long long)source);
+            radio_talker_off(ctx, source);
+        }
+    }
+}
+
 /* Снять говорящих, от которых дольше RADIO_TALKER_TIMEOUT_TB нет кадров:
  * FIN потерян/не был отправлен (пир отвалился, выключил рацию во время разговора).
  * Аналог таймаута источника в radio_audio.c — гарантирует, что индикация «говорит» гаснет. */
@@ -666,7 +687,7 @@ static void radio_flush(struct radio_ctx* ctx, const char* reason) {
     if (!topo_group_data_allowed(ctx->group, NULL)) {
         DEBUG_WARN(DEBUG_CATEGORY_RADIO, "radio: discard revoked batch group=%llx frames=%u",
                    (unsigned long long)ctx->group->group_id, ctx->burst_count);
-        ctx->burst_count = 0; ctx->burst_bytes = 0; ctx->talking = 0; return;
+        radio_revalidate_members(ctx->group); return;
     }
     uint64_t self = ctx->group->instance->node_id;
     uint8_t out[RADIO_HDR_SIZE + 8 + RADIO_MAX_BODY];
@@ -692,7 +713,7 @@ int radio_talk_begin(struct UTUN_INSTANCE* inst, uint64_t group_id) {
     if (!group) { DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: talk_begin group 0x%016llx not found", RADIO_ID, (unsigned long long)group_id); return -1; }
     struct radio_ctx* ctx = radio_of(group);
     if (!ctx) return -1;
-    if (!topo_group_data_allowed(group, NULL)) return -1;
+    if (!topo_group_data_allowed(group, NULL)) { radio_revalidate_members(group); return -1; }
     if (ctx->talking) {
         DEBUG_WARN(DEBUG_CATEGORY_RADIO, "radio: BEGIN closes unfinished stream=%u", ctx->stream_id);
         radio_talk_end(inst, group_id);
@@ -713,7 +734,7 @@ int radio_talk_send(struct UTUN_INSTANCE* inst, uint64_t group_id, const uint8_t
     struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
     struct radio_ctx* ctx = radio_of(group);
     if (!ctx) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: send group unavailable group=%llx", (unsigned long long)group_id); return -1; }
-    if (!topo_group_data_allowed(group, NULL)) { radio_flush(ctx, "revoked"); ctx->talking = 0; return -1; }
+    if (!topo_group_data_allowed(group, NULL)) { radio_revalidate_members(group); return -1; }
     if (!ctx->talking) {
         /* burst не открыт: хвостовой кадр, пришедший после radio_talk_end (гонка релиза PTT) —
          * не переоткрываем передачу, иначе получится burst без FIN и «говорит» зависнет у пиров. */
@@ -744,6 +765,7 @@ int radio_talk_end(struct UTUN_INSTANCE* inst, uint64_t group_id) {
     if (!group) return -1;
     struct radio_ctx* ctx = radio_of(group);
     if (!ctx) return -1;
+    if (!topo_group_data_allowed(group, NULL)) { radio_revalidate_members(group); return -1; }
     if (!ctx->talking) return 0;
 
     radio_flush(ctx, "end");

src/radio/radio.h

@@ -22,6 +22,8 @@ extern "C" {
 
 int  radio_init(struct TOPO_GROUP* group);     /* из topo_group_create */
 void radio_destroy(struct TOPO_GROUP* group);  /* из topo_group_destroy */
+/* После commit членства CHAT: отменить недопущенную локальную передачу и снять индикацию удалённых участников. */
+void radio_revalidate_members(struct TOPO_GROUP* group);
 
 int  radio_init_instance(struct UTUN_INSTANCE* inst);     /* etcp_bind(ETCP_ID_RADIO) */
 void radio_destroy_instance(struct UTUN_INSTANCE* inst);  /* etcp_unbind */

src/routing_layer/topo_group.c

@@ -439,6 +439,7 @@ void topo_group_revalidate_members(struct TOPO_GROUP* group) {
     if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT || group->stopping) return;
     etcp_router_revalidate_group(group->instance, group->group_id);
     conn_mgr_revalidate_members(group->conn_mgr);
+    radio_revalidate_members(group);
     /* Удаление вызывает callbacks и меняет список; каждый раз ищем следующего заново. */
     for (;;) {
         struct TOPO_GROUP_CONN_ITEM* invalid = NULL;