From 3481e2fe692b38ee6f544ebccfcc2e49a6835c5a Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 4 Nov 2026 00:33:40 +0100 Subject: [PATCH] Project member admission inside the caller transaction and notify after commit --- src/chat/group_policy.c | 87 +++++++++++++++++++++++++++++------------ src/chat/group_policy.h | 8 ++++ 2 files changed, 70 insertions(+), 25 deletions(-) diff --git a/src/chat/group_policy.c b/src/chat/group_policy.c index 93e0eedbf..c8c1b5851 100644 --- a/src/chat/group_policy.c +++ b/src/chat/group_policy.c @@ -213,32 +213,55 @@ int group_policy_pending(sqlite3* db, const char* channel) { return rc == SQLITE_ROW; } -int group_policy_refresh(struct UTUN_INSTANCE* inst, const char* channel) { - sqlite3* db = inst ? inst->topo_sqlite_db : NULL; struct group_policy policy; - struct gp_member* m = NULL; size_t count = 0; uint64_t owner; - if (!db) return -1; - int pending = group_policy_pending(db, channel); - if (pending) { - DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "group admission retains previous checkpoint ch=%s pending=%d", channel, pending); - return pending < 0 ? -1 : 0; +struct group_policy_update { + char channel[21]; + struct group_policy policy; + struct gp_member* members; + size_t count; + int changes; +}; + +void group_policy_discard(struct group_policy_update* update) { + if (update) { u_free(update->members); u_free(update); } +} + +/* Проекция допуска сохраняется той же транзакцией, что и породившие её удостоверения. */ +int group_policy_project(struct UTUN_INSTANCE* inst, const char* channel, struct group_policy_update** out) { + sqlite3* db = inst ? inst->topo_sqlite_db : NULL; + if (out) *out = NULL; + if (!db || !out || !chat_channel_id(channel) || strlen(channel) > 20 || sqlite3_get_autocommit(db)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "group admission project requires channel and caller transaction"); return -1; } - if (!db || !chat_channel_id(channel) || !sqlite3_get_autocommit(db) || group_policy_get(db, channel, &policy) || - topo_node_sqlite_channel_get(db, channel, NULL, 0, &owner, NULL, NULL, NULL) || gp_load(inst, channel, &m, &count)) return -1; - if (gp_evaluate(m, count, policy.all_members, owner)) { u_free(m); return -1; } - sqlite3_stmt* st = NULL; char sql[160]; int result = -1, changes = 0; - if (sqlite3_exec(db, "BEGIN IMMEDIATE /* admission */", NULL, NULL, NULL) != SQLITE_OK) goto done; + struct group_policy_update* update = u_calloc(1, sizeof(*update)); + if (!update) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "group admission allocation ch=%s", channel); return -1; } + strcpy(update->channel, channel); + sqlite3_stmt* st = NULL; char sql[192]; uint64_t owner; + if (group_policy_get(db, channel, &update->policy) || + topo_node_sqlite_channel_get(db, channel, NULL, 0, &owner, NULL, NULL, NULL) || + gp_load(inst, channel, &update->members, &update->count) || + gp_evaluate(update->members, update->count, update->policy.all_members, owner)) goto error; snprintf(sql, sizeof(sql), "UPDATE peers_%s SET admitted=? WHERE node_id=? AND record_state=0", channel); - if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) goto done; - for (size_t i = 0; i < count; i++) { + if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) goto error; + struct gp_member* m = update->members; + for (size_t i = 0; i < update->count; i++) { if (m[i].previous == m[i].allowed) continue; if (sqlite3_bind_int(st, 1, m[i].allowed) != SQLITE_OK || sqlite3_bind_int64(st, 2, (sqlite3_int64)m[i].node) != SQLITE_OK || sqlite3_step(st) != SQLITE_DONE || - sqlite3_changes(db) != 1 || sqlite3_reset(st) != SQLITE_OK) goto done; - changes++; + sqlite3_changes(db) != 1 || sqlite3_reset(st) != SQLITE_OK) goto error; + update->changes++; } - { int rc = sqlite3_finalize(st); st = NULL; if (rc != SQLITE_OK) goto done; } - if (sqlite3_exec(db, "COMMIT /* admission */", NULL, NULL, NULL) != SQLITE_OK) goto done; - result = 0; + { int rc = sqlite3_finalize(st); st = NULL; if (rc != SQLITE_OK) goto error; } + *out = update; return 0; +error: + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "group admission projection failed ch=%s: %s", channel, sqlite3_errmsg(db)); + sqlite3_finalize(st); group_policy_discard(update); return -1; +} + +void group_policy_publish(struct UTUN_INSTANCE* inst, struct group_policy_update* update) { + if (!update) return; + const char* channel = update->channel; + struct gp_member* m = update->members; size_t count = update->count; + struct group_policy policy = update->policy; struct TOPO_GROUP* group = inst->topo_groups ? topo_groups_find(inst->topo_groups, chat_channel_id(channel)) : NULL; if (group) topo_group_revalidate_members(group); size_t self = gp_find(m, count, inst->node_id); @@ -254,21 +277,35 @@ int group_policy_refresh(struct UTUN_INSTANCE* inst, const char* channel) { struct ETCP_CONN* conn = instance_find_conn(inst, m[i].node); if (g && conn && m[i].allowed && conn->links_up && !conn->close_requested) topo_group_new_conn(g, conn); } - if (changes) { + if (update->changes) { merkle_sync_revalidate_peers(inst, MT_DATA_FEEDBACK, channel); db_sync_members_changed(inst, channel); uint8_t evt[21]; evt[0] = (uint8_t)strlen(channel); memcpy(evt + 1, channel, evt[0]); chat_event_post(inst, CHAT_EVT_MEMBERS_CHANGED, evt, evt[0] + 1); } -done: - if (result < 0) { + group_policy_discard(update); +} + +int group_policy_refresh(struct UTUN_INSTANCE* inst, const char* channel) { + sqlite3* db = inst ? inst->topo_sqlite_db : NULL; + if (!db || !chat_channel_id(channel) || !sqlite3_get_autocommit(db)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "group admission refresh requires database outside transaction"); return -1; + } + int pending = group_policy_pending(db, channel); + if (pending) { + DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "group admission retains previous checkpoint ch=%s pending=%d", channel, pending); + return pending < 0 ? -1 : 0; + } + struct group_policy_update* update = NULL; + if (sqlite3_exec(db, "BEGIN IMMEDIATE /* admission */", NULL, NULL, NULL) != SQLITE_OK || + group_policy_project(inst, channel, &update) || sqlite3_exec(db, "COMMIT /* admission */", NULL, NULL, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "group admission transaction ch=%s: %s", channel, sqlite3_errmsg(db)); - sqlite3_finalize(st); if (!sqlite3_get_autocommit(db) && sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "group admission rollback failed ch=%s", channel); utun_instance_stop(inst); } + group_policy_discard(update); return -1; } - u_free(m); return result; + group_policy_publish(inst, update); return 0; } int group_policy_can_invite(struct UTUN_INSTANCE* inst, const char* channel, uint64_t inviter) { diff --git a/src/chat/group_policy.h b/src/chat/group_policy.h index b2e7f5d7e..7a9ed7109 100644 --- a/src/chat/group_policy.h +++ b/src/chat/group_policy.h @@ -8,6 +8,7 @@ #include struct UTUN_INSTANCE; struct sqlite3; +struct group_policy_update; #ifdef __cplusplus extern "C" { #endif @@ -30,6 +31,13 @@ int group_policy_decode(struct sqlite3* db, const char* channel, const uint8_t* int group_policy_apply(struct sqlite3* db, const char* channel, const struct group_policy* policy); /* Пересчитать admitted атомарно, затем снять недопущенные сессии; 0/-1. */ int group_policy_refresh(struct UTUN_INSTANCE* inst, const char* channel); +/* Рассчитать и записать допуск В открытой транзакции caller. 0/-1; out принадлежит caller. + * После COMMIT вызвать publish, после ROLLBACK — discard. До publish внешних callbacks нет. */ +int group_policy_project(struct UTUN_INSTANCE* inst, const char* channel, struct group_policy_update** out); +/* Уведомить сервисы после commit и освободить update; callbacks могут удалять группу. */ +void group_policy_publish(struct UTUN_INSTANCE* inst, struct group_policy_update* update); +/* Освободить расчёт без уведомлений (rollback/stop); NULL допустим. */ +void group_policy_discard(struct group_policy_update* update); /* Незавершённый приём страниц: 1/0/-1. Маркер хранится локально, не входит в Merkle. */ int group_policy_pending(struct sqlite3* db, const char* channel); /* Текущее право подписи приглашения; 1/0/-1, ошибка никогда не разрешает действие. */