454 lines
22 KiB
C
454 lines
22 KiB
C
/*
|
|
* dm_mailbox.c — durable очередь суперузла. Сообщения и квитанции непрозрачны;
|
|
* приём/удаление атомарны, роли и область маршрута проверяются на каждом PUT.
|
|
*/
|
|
#include "dm_mailbox.h"
|
|
#include "dm_core.h"
|
|
#include "dm_mailbox_media.h"
|
|
#include "dm_media_priv.h"
|
|
#include "../utun_instance.h"
|
|
#include "../ntp_time.h"
|
|
#include "../routing_layer/etcp_router.h"
|
|
#include "../routing_layer/topo_node_sqlite.h"
|
|
#include "../transport_layer/etcp.h"
|
|
#include "../../lib/mem.h"
|
|
#include "../../lib/ll_queue.h"
|
|
#include "../../lib/debug_config.h"
|
|
#include "../../lib/json_flat.h"
|
|
#include <sqlite3.h>
|
|
#include <string.h>
|
|
#include <stdio.h>
|
|
|
|
#define MB_ID "dm_mailbox"
|
|
#define MB_PUT 1 /* [recipient:8][каноническое тело] */
|
|
#define MB_DELIVER 2 /* [каноническое тело] */
|
|
#define MB_RECEIPT 3 /* [подписанная квитанция получателя] */
|
|
#define MB_STORED 4 /* [conv:8][seq:8] — durable PUT подтверждён суперузлом */
|
|
|
|
struct dm_mb_state {
|
|
struct UTUN_INSTANCE* inst;
|
|
sqlite3* db;
|
|
void* timer;
|
|
struct dm_mb_media* media;
|
|
};
|
|
|
|
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 (!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;
|
|
}
|
|
if (!etcp_router_send_q_has_room(mb->inst, gid, dst, ETCP_RT_ID_DM_MAILBOX)) {
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: backpressure peer=%llu", MB_ID, (unsigned long long)dst);
|
|
return -1;
|
|
}
|
|
struct ll_entry* e = queue_entry_new(0);
|
|
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: packet allocation failed", MB_ID); return -1; }
|
|
e->dgram = u_malloc(len + 2);
|
|
if (!e->dgram) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: payload allocation failed", MB_ID);
|
|
queue_entry_free(e);
|
|
return -1;
|
|
}
|
|
e->dgram[0] = ETCP_RT_ID_DM_MAILBOX;
|
|
e->dgram[1] = cmd;
|
|
memcpy(e->dgram + 2, body, len);
|
|
e->len = (uint16_t)(len + 2);
|
|
int rc = etcp_route_send(mb->inst, gid, dst, e, 1, ROUTER_FLAG_SIGNED);
|
|
if (rc) DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: route failed peer=%llu group=%llu cmd=%u rc=%d", MB_ID,
|
|
(unsigned long long)dst, (unsigned long long)gid, cmd, rc);
|
|
return rc;
|
|
}
|
|
|
|
/* Проверять каждый SQL результат, в том числе границы транзакций. */
|
|
static int mb_exec(struct dm_mb_state* mb, const char* sql) {
|
|
int rc = sqlite3_exec(mb->db, sql, NULL, NULL, NULL);
|
|
if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: SQL rc=%d error=%s statement=%s",
|
|
MB_ID, rc, sqlite3_errmsg(mb->db), sql);
|
|
return rc == SQLITE_OK ? 0 : -1;
|
|
}
|
|
|
|
/* Принять неизменяемое тело; подтверждать только durable запись или точный повтор. */
|
|
static void mb_put_received(struct dm_mb_state* mb, uint64_t gid, uint64_t source, const uint8_t* p, size_t len) {
|
|
if (len < 8) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: short PUT bytes=%zu", MB_ID, len); return; }
|
|
uint64_t recipient, author, seq;
|
|
memcpy(&recipient, p, 8);
|
|
p += 8;
|
|
len -= 8;
|
|
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)) {
|
|
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;
|
|
}
|
|
sqlite3_stmt* st = NULL;
|
|
if (mb_exec(mb, "BEGIN IMMEDIATE") != 0) return;
|
|
if (dm_mailbox_media_register(mb->media, gid, recipient, p, len) != 0) {
|
|
mb_exec(mb, "ROLLBACK");
|
|
return;
|
|
}
|
|
int rc = sqlite3_prepare_v2(mb->db,
|
|
"SELECT receipt FROM dm_mail_receipts WHERE recipient=? AND sender=? AND seq=?", -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);
|
|
rc = sqlite3_step(st);
|
|
if (rc == SQLITE_ROW) {
|
|
uint8_t receipt[DM_RECEIPT_SIZE];
|
|
if (sqlite3_column_bytes(st, 0) != DM_RECEIPT_SIZE) goto fail;
|
|
memcpy(receipt, sqlite3_column_blob(st, 0), sizeof(receipt));
|
|
sqlite3_finalize(st);
|
|
st = NULL;
|
|
if (mb_exec(mb, "COMMIT") != 0) goto fail;
|
|
if (dm_verify_receipt(mb->inst, receipt, p, len) == 0)
|
|
mb_send(mb, gid, author, MB_RECEIPT, receipt, sizeof(receipt));
|
|
return; /* Поздний PUT не возрождает доставленную задачу. */
|
|
}
|
|
if (rc != SQLITE_DONE) goto fail;
|
|
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)"
|
|
" 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);
|
|
st = NULL;
|
|
rc = sqlite3_prepare_v2(mb->db,
|
|
"SELECT msg FROM dm_pending_messages WHERE recipient=? AND sender=? AND seq=?", -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);
|
|
rc = sqlite3_step(st);
|
|
if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != (int)len ||
|
|
memcmp(sqlite3_column_blob(st, 0), p, len)) goto fail;
|
|
sqlite3_finalize(st);
|
|
st = NULL;
|
|
if (mb_exec(mb, "COMMIT") != 0) goto fail;
|
|
mb_send(mb, gid, author, MB_STORED, p, 16);
|
|
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: stored sender=%llu recipient=%llu seq=%llu group=%llu", MB_ID,
|
|
(unsigned long long)author, (unsigned long long)recipient, (unsigned long long)seq, (unsigned long long)gid);
|
|
return;
|
|
fail:
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: PUT database/conflict rc=%d error=%s", MB_ID, rc, sqlite3_errmsg(mb->db));
|
|
sqlite3_finalize(st);
|
|
mb_exec(mb, "ROLLBACK");
|
|
}
|
|
|
|
/* Сначала durable квитанция, затем удаление точной копии в той же транзакции. */
|
|
static void mb_receipt_received(struct dm_mb_state* mb, uint64_t gid, uint64_t source, const uint8_t* r) {
|
|
if (dm_verify_receipt(mb->inst, r, NULL, 0) != 0) return;
|
|
uint64_t author, recipient, seq;
|
|
memcpy(&seq, r + 16, 8);
|
|
memcpy(&author, r + 24, 8);
|
|
memcpy(&recipient, r + 32, 8);
|
|
if (author == mb->inst->node_id) {
|
|
dm_accept_receipt(mb->inst, r);
|
|
return;
|
|
}
|
|
if ((source != author && source != recipient) || !mb_scope(mb, gid, mb->inst->node_id, author, 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;
|
|
}
|
|
if (mb_exec(mb, "BEGIN IMMEDIATE") != 0) return;
|
|
sqlite3_stmt* st = NULL;
|
|
int rc = sqlite3_prepare_v2(mb->db,
|
|
"SELECT msg FROM dm_pending_messages WHERE recipient=? AND sender=? AND seq=?", -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);
|
|
rc = sqlite3_step(st);
|
|
if (rc == SQLITE_ROW) {
|
|
if (dm_verify_receipt(mb->inst, r, sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0)) != 0) goto fail;
|
|
} else if (rc != SQLITE_DONE) goto fail;
|
|
sqlite3_finalize(st);
|
|
st = NULL;
|
|
rc = sqlite3_prepare_v2(mb->db,
|
|
"INSERT INTO dm_mail_receipts(recipient,sender,seq,receipt,group_id) 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);
|
|
st = NULL;
|
|
rc = sqlite3_prepare_v2(mb->db, "DELETE FROM dm_pending_messages WHERE recipient=? AND sender=? AND seq=?", -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);
|
|
rc = sqlite3_step(st);
|
|
if (rc != SQLITE_DONE) goto fail;
|
|
int deleted = sqlite3_changes(mb->db);
|
|
sqlite3_finalize(st);
|
|
st = NULL;
|
|
if (mb_exec(mb, "COMMIT") != 0) goto fail;
|
|
if (deleted) DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: delivered copy removed sender=%llu recipient=%llu seq=%llu", MB_ID,
|
|
(unsigned long long)author, (unsigned long long)recipient, (unsigned long long)seq);
|
|
mb_send(mb, gid, author, MB_RECEIPT, r, DM_RECEIPT_SIZE);
|
|
return;
|
|
fail:
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receipt commit failed rc=%d error=%s", MB_ID, rc, sqlite3_errmsg(mb->db));
|
|
sqlite3_finalize(st);
|
|
mb_exec(mb, "ROLLBACK");
|
|
}
|
|
|
|
/* Никаких диапазонных ACK: одна квитанция подтверждает одно тело. */
|
|
static void mb_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
|
|
if (!conn || !entry || !entry->dgram || entry->len <= ROUTER_SVC_PAYLOAD_OFF) {
|
|
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
|
|
return;
|
|
}
|
|
struct dm_mb_state* mb = conn->instance->dm_mailbox;
|
|
if (!mb) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: receive after stop", MB_ID); goto done; }
|
|
uint64_t gid, source;
|
|
memcpy(&gid, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
|
|
memcpy(&source, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
|
|
const uint8_t* p = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
|
|
size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF - 1;
|
|
if (!(entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unsigned control source=%llu", MB_ID, (unsigned long long)source);
|
|
goto done;
|
|
}
|
|
if (p[0] == MB_PUT) mb_put_received(mb, gid, source, p + 1, len);
|
|
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)) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: delivery from non-supernode source=%llu", MB_ID, (unsigned long long)source);
|
|
goto done;
|
|
}
|
|
uint8_t receipt[DM_RECEIPT_SIZE];
|
|
if (dm_accept_message(mb->inst, p + 1, len, receipt) == 0)
|
|
mb_send(mb, gid, source, MB_RECEIPT, receipt, sizeof(receipt));
|
|
} else if (p[0] == MB_RECEIPT && len == DM_RECEIPT_SIZE) mb_receipt_received(mb, gid, source, p + 1);
|
|
else if (p[0] == MB_STORED && len == 16) {
|
|
uint64_t conv, seq;
|
|
memcpy(&conv, p + 1, 8);
|
|
memcpy(&seq, p + 9, 8);
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: custody confirmed super=%llu conv=%llu seq=%llu (await recipient)", MB_ID,
|
|
(unsigned long long)source, (unsigned long long)conv, (unsigned long long)seq);
|
|
} else DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: invalid command=%u bytes=%zu", MB_ID, p[0], len);
|
|
done:
|
|
queue_dgram_free(entry);
|
|
queue_entry_free(entry);
|
|
}
|
|
|
|
/* Продвигать ограниченную порцию; SQLite statement закрыт до вызова транспорта. */
|
|
static void mb_tick(void* arg) {
|
|
struct dm_mb_state* mb = arg;
|
|
mb->timer = NULL;
|
|
int64_t now = (int64_t)ntp_time_get_seconds(mb->inst);
|
|
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"
|
|
" 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);
|
|
rc = sqlite3_step(st);
|
|
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);
|
|
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; }
|
|
memcpy(msg, sqlite3_column_blob(st, 3), len);
|
|
sqlite3_finalize(st);
|
|
st = NULL;
|
|
rc = sqlite3_prepare_v2(mb->db,
|
|
"UPDATE dm_pending_messages SET retry_at=? WHERE recipient=? AND sender=? AND seq=?", -1, &st, NULL);
|
|
if (rc == SQLITE_OK) {
|
|
sqlite3_bind_int64(st, 1, now + 5);
|
|
sqlite3_bind_int64(st, 2, (sqlite3_int64)recipient);
|
|
sqlite3_bind_int64(st, 3, (sqlite3_int64)author);
|
|
sqlite3_bind_int64(st, 4, (sqlite3_int64)seq);
|
|
rc = sqlite3_step(st);
|
|
}
|
|
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)) {
|
|
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,
|
|
(unsigned long long)route_gid);
|
|
}
|
|
u_free(msg);
|
|
continue;
|
|
fail:
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: pending scan rc=%d error=%s", MB_ID, rc, sqlite3_errmsg(mb->db));
|
|
sqlite3_finalize(st);
|
|
break;
|
|
}
|
|
mb->timer = uasync_set_timeout(mb->inst->ua, 10000, mb, mb_tick, "dm_mailbox");
|
|
if (!mb->timer) DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: retry timer allocation failed", MB_ID);
|
|
}
|
|
|
|
/* Новые таблицы протокола; прежний формат и диапазонные ACK не используются. */
|
|
int dm_mailbox_init(struct UTUN_INSTANCE* inst) {
|
|
if (!inst || !inst->topo_sqlite_db) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: init without core DB", MB_ID); return -1; }
|
|
if (inst->dm_mailbox) return 0;
|
|
struct dm_mb_state* mb = u_calloc(1, sizeof(*mb));
|
|
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;
|
|
inst->dm_mailbox = mb;
|
|
mb->media = dm_mailbox_media_create(inst);
|
|
if (!mb->media) {
|
|
etcp_router_unbind(inst, ETCP_RT_ID_DM_MAILBOX);
|
|
inst->dm_mailbox = NULL;
|
|
goto fail;
|
|
}
|
|
mb->timer = uasync_set_timeout(inst->ua, 10000, mb, mb_tick, "dm_mailbox");
|
|
if (!mb->timer) {
|
|
etcp_router_unbind(inst, ETCP_RT_ID_DM_MAILBOX);
|
|
inst->dm_mailbox = NULL;
|
|
goto fail;
|
|
}
|
|
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: initialized persistent supernode queue", MB_ID);
|
|
return 0;
|
|
fail:
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: initialization failed", MB_ID);
|
|
dm_mailbox_media_destroy(mb->media);
|
|
u_free(mb);
|
|
return -1;
|
|
}
|
|
|
|
/* Отменить таймер до освобождения owner, durable копии оставить в БД. */
|
|
void dm_mailbox_destroy(struct UTUN_INSTANCE* inst) {
|
|
if (!inst || !inst->dm_mailbox) return;
|
|
struct dm_mb_state* mb = inst->dm_mailbox;
|
|
if (mb->timer) uasync_cancel_timeout(inst->ua, mb->timer);
|
|
dm_mailbox_media_destroy(mb->media);
|
|
etcp_router_unbind(inst, ETCP_RT_ID_DM_MAILBOX);
|
|
inst->dm_mailbox = NULL;
|
|
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: destroyed, pending retained", MB_ID);
|
|
u_free(mb);
|
|
}
|
|
|
|
/* Найти суперузлы в общей группе обеих сторон; для PUT выбрать только один. */
|
|
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;
|
|
int count = 0, sent = 0;
|
|
if (topo_node_sqlite_get_member_channels(mb->db, mb->inst->node_id, &ids, &count) != 0) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: membership lookup failed", MB_ID);
|
|
return -1;
|
|
}
|
|
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);
|
|
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));
|
|
continue;
|
|
}
|
|
int rc;
|
|
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
|
|
uint64_t node = (uint64_t)sqlite3_column_int64(st, 0);
|
|
const char* tags = (const char*)sqlite3_column_text(st, 1);
|
|
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 (!mb_send(mb, ids[i], node, cmd, body, len)) {
|
|
sent++;
|
|
if (cmd == MB_PUT) break;
|
|
}
|
|
}
|
|
if (rc != SQLITE_ROW && rc != SQLITE_DONE)
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: supernode scan failed rc=%d", MB_ID, rc);
|
|
sqlite3_finalize(st);
|
|
if (sent && cmd == MB_PUT) break;
|
|
}
|
|
u_free(ids);
|
|
return sent;
|
|
}
|
|
|
|
/* Исходящее тело остаётся у A до подписанного подтверждения B. */
|
|
int dm_mailbox_put(struct UTUN_INSTANCE* inst, uint64_t recipient, uint64_t sender, const uint8_t* msg, size_t len) {
|
|
struct dm_mb_state* mb = inst ? inst->dm_mailbox : NULL;
|
|
if (!mb || sender != inst->node_id || dm_verify_message(inst, recipient, msg, len) != 0) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: invalid local PUT", MB_ID);
|
|
return -1;
|
|
}
|
|
uint8_t body[8 + DM_MEDIA_BODY_MAX];
|
|
memcpy(body, &recipient, 8);
|
|
memcpy(body + 8, msg, len);
|
|
int n = mb_to_supernodes(mb, recipient, sender, MB_PUT, body, len + 8);
|
|
if (n <= 0) DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: no reachable supernode; local outbox retained recipient=%llu", MB_ID,
|
|
(unsigned long long)recipient);
|
|
return n > 0 ? 0 : -1;
|
|
}
|
|
|
|
/* Копии могут остаться после смены маршрута/суперузла — удалить каждую доступную. */
|
|
void dm_mailbox_receipt(struct UTUN_INSTANCE* inst, const uint8_t* receipt) {
|
|
struct dm_mb_state* mb = inst ? inst->dm_mailbox : NULL;
|
|
if (!mb || dm_verify_receipt(inst, receipt, NULL, 0) != 0) return;
|
|
uint64_t author, recipient;
|
|
memcpy(&author, receipt + 24, 8);
|
|
memcpy(&recipient, receipt + 32, 8);
|
|
mb_to_supernodes(mb, recipient, author, MB_RECEIPT, receipt, DM_RECEIPT_SIZE);
|
|
}
|