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

dm: route mailbox delivery by each pair of node identities

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 22:13:29 +01:00
1 parent 3c58fc80e3
commit 29013ebccc
3 files changed
+50 -56
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_mailbox.c
+28 -52
View File
@@ -10,6 +10,7 @@
#include "../ntp_time.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_group.h"
#include "../transport_layer/etcp.h"
#include "../../lib/mem.h"
#include "../../lib/ll_queue.h"
@@ -37,33 +38,11 @@ struct dm_mb_media* dm_mailbox_get_media(struct UTUN_INSTANCE* inst) {
return inst && inst->dm_mailbox ? inst->dm_mailbox->media : NULL;
}
/* Проверить роль в конкретной общей группе; storage не заменяет supernode. */
static int mb_scope(struct dm_mb_state* mb, uint64_t gid, uint64_t super, uint64_t a, uint64_t b) {
if (!gid) return 0;
char ch[64];
snprintf(ch, sizeof(ch), "%llu", (unsigned long long)gid);
char sql[256];
snprintf(sql, sizeof(sql), "SELECT adm_tags FROM \"peers_%s\" WHERE node_id=? AND deleted=0", ch);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(mb->db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: role query failed group=%llu: %s", MB_ID,
(unsigned long long)gid, sqlite3_errmsg(mb->db));
return 0;
}
sqlite3_bind_int64(st, 1, (sqlite3_int64)super);
int allowed = 0;
char role[16];
if (sqlite3_step(st) == SQLITE_ROW) {
const char* tags = (const char*)sqlite3_column_text(st, 0);
allowed = tags && json_flat_get(tags, "supernode", role, sizeof(role)) == 0 && !strcmp(role, "yes");
}
sqlite3_finalize(st);
return allowed && topo_node_sqlite_member_in_channel(mb->db, ch, a) &&
topo_node_sqlite_member_in_channel(mb->db, ch, b);
}
/* Подписать контроль и передать владение пакетом router, учитывая backpressure. */
static int mb_send(struct dm_mb_state* mb, uint64_t gid, uint64_t dst, uint8_t cmd, const uint8_t* body, size_t len) {
if (!dm_route_group(mb->inst, dst, gid, &gid)) {
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: no admitted route peer=%llu cmd=%u", MB_ID, (unsigned long long)dst, cmd); return -1;
}
if (!etcp_router_conn_get(mb->inst, gid, dst, ETCP_RT_ID_DM_MAILBOX)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: router channel allocation failed peer=%llu", MB_ID, (unsigned long long)dst);
return -1;
@@ -108,7 +87,7 @@ static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t sourc
if (dm_verify_message(mb->inst, recipient, p, len) != 0) return;
memcpy(&seq, p + 8, 8);
memcpy(&author, p + 24, 8);
if (source != author || !mb_scope(mb, gid, mb->inst->node_id, author, recipient)) {
if (source != author || !dm_peer_role(mb->inst, mb->inst->node_id, "supernode")) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unauthorized PUT source=%llu recipient=%llu group=%llu", MB_ID,
(unsigned long long)source, (unsigned long long)recipient, (unsigned long long)gid);
return;
@@ -142,14 +121,13 @@ static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t sourc
sqlite3_finalize(st);
st = NULL;
rc = sqlite3_prepare_v2(mb->db,
"INSERT INTO dm_pending_messages(recipient,sender,seq,msg,group_id,retry_at) VALUES(?,?,?,?,?,0)"
"INSERT INTO dm_pending_messages(recipient,sender,seq,msg,retry_at) VALUES(?,?,?,?,0)"
" ON CONFLICT(recipient,sender,seq) DO NOTHING", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_int64(st, 1, (sqlite3_int64)recipient);
sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
sqlite3_bind_blob(st, 4, p, (int)len, SQLITE_STATIC);
sqlite3_bind_int64(st, 5, (sqlite3_int64)gid);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
sqlite3_finalize(st);
@@ -188,7 +166,7 @@ static void mb_receipt_received(struct dm_mb_state* mb, uint64_t gid, uint64_t s
dm_accept_receipt(mb->inst, r);
return;
}
if ((source != author && source != recipient) || !mb_scope(mb, gid, mb->inst->node_id, author, recipient)) {
if (source != author && source != recipient) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: receipt outside authorized scope source=%llu group=%llu", MB_ID,
(unsigned long long)source, (unsigned long long)gid);
return;
@@ -208,14 +186,13 @@ static void mb_receipt_received(struct dm_mb_state* mb, uint64_t gid, uint64_t s
sqlite3_finalize(st);
st = NULL;
rc = sqlite3_prepare_v2(mb->db,
"INSERT INTO dm_mail_receipts(recipient,sender,seq,receipt,group_id) VALUES(?,?,?,?,?)"
"INSERT INTO dm_mail_receipts(recipient,sender,seq,receipt) VALUES(?,?,?,?)"
" ON CONFLICT(recipient,sender,seq) DO NOTHING", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_int64(st, 1, (sqlite3_int64)recipient);
sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
sqlite3_bind_blob(st, 4, r, DM_RECEIPT_SIZE, SQLITE_STATIC);
sqlite3_bind_int64(st, 5, (sqlite3_int64)gid);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
sqlite3_finalize(st);
@@ -258,12 +235,14 @@ static void mb_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unsigned control source=%llu", MB_ID, (unsigned long long)source);
goto done;
}
if (!topo_group_chat_allowed(mb->inst, gid, source)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: denied group=%llu source=%llu cmd=%u", MB_ID,
(unsigned long long)gid, (unsigned long long)source, p[0]); goto done;
}
if (p[0] == MB_PUT || p[0] == MB_MEDIA_REQUEST) mb_put_received(mb, gid, source, p + 1, len, p[0] == MB_MEDIA_REQUEST);
else if (p[0] == MB_DELIVER) {
if (len < DM_MSG_FIXED_HDR) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: short delivery", MB_ID); goto done; }
uint64_t author;
memcpy(&author, p + 25, 8);
if (!mb_scope(mb, gid, source, author, mb->inst->node_id)) {
if (!dm_peer_role(mb->inst, source, "supernode")) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: delivery from non-supernode source=%llu", MB_ID, (unsigned long long)source);
goto done;
}
@@ -276,7 +255,7 @@ static void mb_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
memcpy(&conv, p + 1, 8);
memcpy(&seq, p + 9, 8);
memcpy(&recipient, p + 17, 8);
if (!mb_scope(mb, gid, source, mb->inst->node_id, recipient)) {
if (!dm_peer_role(mb->inst, source, "supernode")) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM mailbox: unauthorized STORED super=%llu", (unsigned long long)source); goto done;
}
char id[64]; snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
@@ -303,7 +282,7 @@ static void mb_tick(void* arg) {
for (int i = 0; i < 32; i++) {
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(mb->db,
"SELECT recipient,sender,seq,msg,group_id FROM dm_pending_messages"
"SELECT recipient,sender,seq,msg FROM dm_pending_messages"
" WHERE retry_at<=? ORDER BY retry_at,seq LIMIT 1", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_int64(st, 1, now);
@@ -311,7 +290,7 @@ static void mb_tick(void* arg) {
if (rc == SQLITE_DONE) { sqlite3_finalize(st); break; }
if (rc != SQLITE_ROW) goto fail;
uint64_t recipient = (uint64_t)sqlite3_column_int64(st, 0), author = (uint64_t)sqlite3_column_int64(st, 1);
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 2), gid = (uint64_t)sqlite3_column_int64(st, 4);
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 2);
size_t len = (size_t)sqlite3_column_bytes(st, 3);
uint8_t* msg = u_malloc(len);
if (!msg) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: delivery allocation failed", MB_ID); sqlite3_finalize(st); break; }
@@ -330,9 +309,8 @@ static void mb_tick(void* arg) {
sqlite3_finalize(st);
st = NULL;
if (rc != SQLITE_DONE) { u_free(msg); goto fail; }
uint64_t route_gid = gid;
if (dm_route_group(mb->inst, recipient, gid, &route_gid) &&
mb_scope(mb, route_gid, mb->inst->node_id, author, recipient)) {
uint64_t route_gid = 0;
if (dm_route_group(mb->inst, recipient, 0, &route_gid)) {
mb_send(mb, route_gid, recipient, MB_DELIVER, msg, len);
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: pending delivery sender=%llu recipient=%llu seq=%llu group=%llu", MB_ID,
(unsigned long long)author, (unsigned long long)recipient, (unsigned long long)seq,
@@ -357,13 +335,12 @@ int dm_mailbox_init(struct UTUN_INSTANCE* inst) {
if (!mb) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: state allocation failed", MB_ID); return -1; }
mb->inst = inst;
mb->db = inst->topo_sqlite_db;
if (mb_exec(mb, "CREATE TABLE IF NOT EXISTS dm_pending_messages (recipient INTEGER NOT NULL,sender INTEGER NOT NULL,"
"seq INTEGER NOT NULL,msg BLOB NOT NULL,group_id INTEGER NOT NULL,retry_at INTEGER NOT NULL DEFAULT 0,"
"PRIMARY KEY(recipient,sender,seq))") != 0 ||
mb_exec(mb, "CREATE TABLE IF NOT EXISTS dm_mail_receipts (recipient INTEGER NOT NULL,sender INTEGER NOT NULL,"
"seq INTEGER NOT NULL,receipt BLOB NOT NULL,group_id INTEGER NOT NULL,"
"PRIMARY KEY(recipient,sender,seq))") != 0) goto fail;
if (etcp_router_bind(inst, ETCP_RT_ID_DM_MAILBOX, mb_recv) != 0) goto fail;
if (etcp_router_require_crypto(inst, ETCP_RT_ID_DM_MAILBOX, ROUTER_FLAG_SIGNED) ||
etcp_router_set_authorize(inst, ETCP_RT_ID_DM_MAILBOX, topo_group_chat_allowed)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: router policy setup failed", MB_ID);
etcp_router_unbind(inst, ETCP_RT_ID_DM_MAILBOX); goto fail;
}
inst->dm_mailbox = mb;
mb->media = dm_mailbox_media_create(inst);
if (!mb->media) {
@@ -398,7 +375,7 @@ void dm_mailbox_destroy(struct UTUN_INSTANCE* inst) {
u_free(mb);
}
/* Найти суперузлы в общей группе обеих сторон; для PUT выбрать только один. */
/* Объявления S ищутся среди своих групп, маршрут до выбранного S выбирается отдельно. */
static int mb_to_supernodes(struct dm_mb_state* mb, uint64_t recipient, uint64_t author,
uint8_t cmd, const uint8_t* body, size_t len) {
uint64_t* ids = NULL;
@@ -410,9 +387,9 @@ static int mb_to_supernodes(struct dm_mb_state* mb, uint64_t recipient, uint64_t
for (int i = 0; i < count; i++) {
char ch[64], sql[256];
snprintf(ch, sizeof(ch), "%llu", (unsigned long long)ids[i]);
if (!topo_node_sqlite_member_in_channel(mb->db, ch, author) ||
!topo_node_sqlite_member_in_channel(mb->db, ch, recipient)) continue;
snprintf(sql, sizeof(sql), "SELECT node_id,adm_tags FROM \"peers_%s\" WHERE deleted=0 ORDER BY node_id", ch);
if (!topo_group_chat_allowed(mb->inst, ids[i], mb->inst->node_id)) continue;
snprintf(sql, sizeof(sql), "SELECT node_id,adm_tags FROM \"peers_%s\""
" WHERE source=0 AND deleted=0 AND admitted=1 ORDER BY node_id", ch);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(mb->db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: supernode query failed: %s", MB_ID, sqlite3_errmsg(mb->db));
@@ -425,8 +402,7 @@ static int mb_to_supernodes(struct dm_mb_state* mb, uint64_t recipient, uint64_t
char role[16];
if (!tags || json_flat_get(tags, "supernode", role, sizeof(role)) != 0 || strcmp(role, "yes")) continue;
if (node == mb->inst->node_id || node == recipient || node == author) continue;
struct ETCP_ROUTER_PATH path = etcp_router_get_path(mb->inst, ids[i], node, ETCP_RT_ID_DM_MAILBOX);
if (!path.next_hop_node_id) continue;
if (!topo_group_chat_allowed(mb->inst, ids[i], node)) continue;
if (!mb_send(mb, ids[i], node, cmd, body, len)) {
sent++;
if (cmd == MB_PUT || cmd == MB_MEDIA_REQUEST) break;
src/dm/dm_mailbox.h
+8 -4
View File
@@ -1,12 +1,16 @@
/*
* dm_mailbox.h — очередь offline-сообщений на суперузлах
*
* Суперузел общей группы хранит непрозрачное подписанное тело и повторяет
* Суперузел хранит непрозрачное подписанное тело и повторяет
* доставку при появлении маршрута до получателя, независимо от отправителя.
* Копия удаляется только по квитанции получателя после commit в dm_core.
* Квитанция сохраняется отдельно для повторов PUT и доставки отправителю.
*
* Роль S — supernode=yes в подписанных adm_tags общей CHAT-группы A/B; storage её не заменяет.
* Роль S — supernode=yes в подписанных adm_tags допущенной общей группы проверяющего узла и S.
* Пары A/S и S/B выбирают маршруты независимо; общая группа A/S/B не требуется.
* Каждая передача проверяет текущий допуск self/peer и подпись router, включая controls.
* Группа не записывается в pending/receipt; отзыв пути сохраняет очередь и позволяет другой маршрут.
* storage не заменяет роль supernode. Квитанция точного сообщения завершает прежнюю работу и после отзыва роли S.
* PUT фиксирует сообщение; независимый MEDIA_REQUEST регистрирует задание dm_mailbox_media.
* Сервис ETCP_RT_ID_DM_MAILBOX (0x35); все пакеты подписаны router, числа native-endian:
* PUT=1 [recipient:8][message]; DELIVER=2 [message]; RECEIPT=3 [receipt:136]; STORED=4 [conv:8][seq:8][recipient:8].
@@ -29,13 +33,13 @@ extern "C" {
#endif
/* Все API в uasync. init после chat_core_init, с живыми SQLite/ua/router ядра:
* создаёт таблицы, media-координатор и таймер; 0=успех (также повторный init), -1=ошибка.
* таблицы уже созданы dm_core_init; создаёт media-координатор и таймер; 0=успех, -1=ошибка.
* destroy отменяет таймеры/bind до освобождения owner, сохраняет pending/receipts в SQLite.
* Вызывать после остановки dm_media, до chat_core_destroy; inst=NULL допустим для destroy. */
int dm_mailbox_init(struct UTUN_INSTANCE* inst);
void dm_mailbox_destroy(struct UTUN_INSTANCE* inst);
/* Поставить PUT выбранному доступному суперузлу общей с recipient группы.
/* Поставить PUT выбранному доступному суперузлу общей с локальным автором группы.
* 0 означает принятие router, хранение подтверждается отдельно на проводе.
* До квитанции получателя источник сохраняет задачу в durable outbox.
* sender должен совпадать с inst->node_id; msg/len — полное подписанное тело dm_core.
src/dm/dm_schema.c
+14
View File
@@ -147,6 +147,20 @@ int dm_schema_init(struct UTUN_INSTANCE* inst) {
if (rc == SQLITE_OK) rc = sqlite3_step(st);
final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) { schema_error(db, "retain DEVICE outbox recipients"); goto fail; }
if (schema_exec(db, "CREATE TABLE IF NOT EXISTS dm_pending_messages(recipient INTEGER NOT NULL,sender INTEGER NOT NULL,"
"seq INTEGER NOT NULL,msg BLOB NOT NULL,retry_at INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(recipient,sender,seq));"
"CREATE TABLE IF NOT EXISTS dm_mail_receipts(recipient INTEGER NOT NULL,sender INTEGER NOT NULL,"
"seq INTEGER NOT NULL,receipt BLOB NOT NULL,PRIMARY KEY(recipient,sender,seq))")) goto fail;
const char* mailbox_tables[] = {"dm_pending_messages", "dm_mail_receipts"};
for (unsigned i = 0; i < 2; i++) {
found = schema_has_column(db, mailbox_tables[i], "group_id");
if (found < 0) goto fail;
if (found) {
char sql[128]; snprintf(sql, sizeof(sql), "ALTER TABLE %s DROP COLUMN group_id", mailbox_tables[i]);
if (schema_exec(db, sql)) goto fail;
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM mailbox schema: retained %s, removed route binding", mailbox_tables[i]);
}
}
if (schema_exec(db, "RELEASE dm_schema")) goto fail;
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM schema: ready; author sequence, frozen parties and per-device delivery");
return 0;