jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Project member admission inside the caller transaction and notify after commit

Browse Source
This commit is contained in:
evgeny committed 2026-11-04 00:33:40 +01:00
1 parent 81bd4ae982
commit 3481e2fe69
2 files changed
+70 -25
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/group_policy.c
+62 -25
View File
@@ -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) {
src/chat/group_policy.h
+8
View File
@@ -8,6 +8,7 @@
#include <stdint.h>
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, ошибка никогда не разрешает действие. */