Drain outgoing media tasks before deleting their channel

This commit is contained in:
evgeny committed 2026-10-22 14:45:29 +02:00
1 parent b3a223521c
commit 32d80110cb
2 files changed
+53 -15

No files matched your search

+4
View File
@@ -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) ── */
+49 -15
View File
@@ -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);