From 32d80110cbbbe47fc0f905ca6eec9569b0bd7fc0 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 14:45:29 +0200 Subject: [PATCH] Drain outgoing media tasks before deleting their channel --- src/chat/chat_core_priv.h | 4 +++ src/chat/chat_msg.c | 64 ++++++++++++++++++++++++++++++--------- 2 files changed, 53 insertions(+), 15 deletions(-) diff --git a/src/chat/chat_core_priv.h b/src/chat/chat_core_priv.h index 819aa918..bc45b2ad 100644 --- a/src/chat/chat_core_priv.h +++ b/src/chat/chat_core_priv.h @@ -79,6 +79,7 @@ struct ms_props_cbk; /* member_sync.c */ struct media_index_result; struct ms_apply_cbk; /* member_sync.c */ struct group_avatar; +struct media_submit_ctx; struct chat_core_ctx { struct UTUN_INSTANCE* inst; @@ -92,6 +93,7 @@ struct chat_core_ctx { struct ms_apply_cbk* apply_cbks; /* was member_sync.c g_apply_cbks */ struct group_avatar* group_avatar; uint64_t media_cache_bytes; /* was chat_msg.c g_media_cache_bytes */ + struct media_submit_ctx* media_submissions; /* исходящие задачи до последнего callback; удаление канала ждёт их */ void* media_backfill_timer;/* was chat_msg.c g_media_startup_backfill_timer */ int media_backfill_ran; /* was chat_msg.c g_media_startup_backfill_ran */ struct UTUN_INSTANCE* media_backfill_inst; /* was chat_msg.c g_media_startup_backfill_inst */ @@ -125,6 +127,8 @@ void on_msg_inserted(struct DB_SYNC_INSTANCE* si, const uint8_t message_id[32], int chat_msg_parse_media_body(const char* body, struct media_index_result* result); /* Учесть также расшифрованные производные файлы и применить общую квоту кеша. */ void chat_media_cache_enforce(struct UTUN_INSTANCE* inst); +/* Пометить текущие исходящие задачи канала; 1 — удаление продолжит последний completion, 0 — задач нет. */ +int chat_media_defer_channel_delete(struct UTUN_INSTANCE* inst, const char* channel); /* ── применить настройку group_autoconnect ко всем CHAT-группам (chat_channel.c) ── */ diff --git a/src/chat/chat_msg.c b/src/chat/chat_msg.c index 6f87034b..32853745 100644 --- a/src/chat/chat_msg.c +++ b/src/chat/chat_msg.c @@ -157,6 +157,8 @@ void chat_core_submit_trampoline(void* arg) { /* ─── Медиа-сообщения: async регистрация в media_files + отправка ─── */ struct media_submit_ctx { + struct media_submit_ctx* next; + int delete_pending; struct UTUN_INSTANCE* inst; char channel_id[64]; char content_type[32]; @@ -173,8 +175,32 @@ struct media_submit_ctx { /* Завершение владеющей асинхронной отправки, включая все ошибки подготовки/регистрации. */ static void media_submit_finish(struct media_submit_ctx* ctx, int success) { + struct UTUN_INSTANCE* inst = ctx->inst; + struct chat_core_ctx* cc = CC(inst); + char channel[64]; snprintf(channel, sizeof(channel), "%s", ctx->channel_id); + int delete_pending = ctx->delete_pending; + struct media_submit_ctx** slot = &cc->media_submissions; + while (*slot && *slot != ctx) slot = &(*slot)->next; + if (*slot) *slot = ctx->next; + else DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media submission missing from owner ch=%s", channel); chat_event_submitted(ctx->inst, ctx->request_id, success); u_free(ctx); + if (delete_pending) { + for (struct media_submit_ctx* pending = cc->media_submissions; pending; pending = pending->next) + if (!strcmp(pending->channel_id, channel)) return; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "media submissions drained; continuing channel deletion ch=%s", channel); + chat_core_delete_channel(inst, channel); + } +} + +int chat_media_defer_channel_delete(struct UTUN_INSTANCE* inst, const char* channel) { + struct chat_core_ctx* cc = CC(inst); int pending = 0; + for (struct media_submit_ctx* ctx = cc->media_submissions; ctx; ctx = ctx->next) { + if (strcmp(ctx->channel_id, channel)) continue; + ctx->delete_pending = 1; pending++; + } + if (pending) DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "channel deletion waiting for outgoing media ch=%s tasks=%d", channel, pending); + return pending != 0; } static void on_media_registered(void* arg, int err, const struct media_index_result* result) { @@ -187,6 +213,10 @@ static void on_media_registered(void* arg, int err, const struct media_index_res media_submit_finish(mctx, 0); return; } + if (mctx->delete_pending) { + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "prepared media retained for pending deletion ch=%s", mctx->channel_id); + media_submit_finish(mctx, 0); return; + } struct DB_SYNC_INSTANCE* si = si_find(mctx->inst, mctx->channel_id); if (!si) { @@ -326,14 +356,8 @@ static void on_video_prepared(void* arg, int err, const struct video_info* vi) { CC_ID, vi->width, vi->height, (long long)vi->duration_ms); } - char media_base[1024]; - { - const char* last_slash = strrchr(cc->db_path, '/'); - if (last_slash) - snprintf(media_base, sizeof(media_base), "%.*s", (int)(last_slash - cc->db_path), cc->db_path); - else - snprintf(media_base, sizeof(media_base), "%s", cc->db_path); - } + char media_base[MEDIA_BASE_MAX]; + if (media_index_get_base(cc->db, media_base)) { media_submit_finish(mctx, 0); return; } media_index_register_async( mctx->inst->media_async, mctx->inst->ua, cc->db, @@ -351,6 +375,18 @@ void chat_core_submit_media_message(struct UTUN_INSTANCE* inst, struct chat_msg_ if (req) chat_event_submitted(inst, req->request_id, 0); return; } + for (struct media_submit_ctx* pending = cc->media_submissions; pending; pending = pending->next) { + if ((!strcmp(pending->channel_id, req->channel_id) && pending->delete_pending) + || !strcmp(pending->media_dest, req->media_dest)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "media submit rejected: channel deleting/destination busy ch=%s path=%s", + req->channel_id, req->media_dest); + chat_event_submitted(inst, req->request_id, 0); return; + } + } + if (!si_find(inst, req->channel_id)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media submit without channel ch=%s", req->channel_id); + chat_event_submitted(inst, req->request_id, 0); return; + } char reply_hex[65]; if (msg_reply_hex(cc, req->channel_id, req->reply_to, reply_hex) < 0) { chat_event_submitted(inst, req->request_id, 0); return; @@ -383,13 +419,10 @@ void chat_core_submit_media_message(struct UTUN_INSTANCE* inst, struct chat_msg_ } /* build media_base from db_path: dirname(db_path) e.g. /path/to/data */ - char media_base[1024]; - { - const char* last_slash = strrchr(cc->db_path, '/'); - if (last_slash) - snprintf(media_base, sizeof(media_base), "%.*s", (int)(last_slash - cc->db_path), cc->db_path); - else - snprintf(media_base, sizeof(media_base), "%s", cc->db_path); + char media_base[MEDIA_BASE_MAX], location[MEDIA_PATH_MAX]; + if (media_index_get_base(cc->db, media_base) || media_index_location(req->media_dest, media_base, location) + || media_index_track_preparation(cc->db, req->channel_id, req->media_dest, media_base)) { + chat_event_submitted(inst, req->request_id, 0); return; } /* copy fields for async callback (req is freed by trampoline immediately) */ @@ -400,6 +433,7 @@ void chat_core_submit_media_message(struct UTUN_INSTANCE* inst, struct chat_msg_ } mctx->inst = inst; mctx->request_id = req->request_id; + mctx->next = cc->media_submissions; cc->media_submissions = mctx; memcpy(mctx->reply_hex, reply_hex, strlen(reply_hex) + 1); snprintf(mctx->channel_id, sizeof(mctx->channel_id), "%s", req->channel_id); snprintf(mctx->content_type, sizeof(mctx->content_type), "%s", req->content_type);