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

Deliver immutable messages per device with ACK deadline fallback

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 12:36:14 +01:00
1 parent 6c4e6254dd
commit 1b87f3efe9
1 file changed
+86 -18
Download Patch File Download Diff File Expand all files Collapse all files
src/dm/dm_core.c
+86 -18
View File
@@ -657,16 +657,19 @@ int dm_verify_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEI
memcpy(&seq, receipt + 16, 8);
memcpy(&author, receipt + 24, 8);
memcpy(&recipient, receipt + 32, 8);
if (!author || !recipient || author == recipient || !seq || seq > INT64_MAX ||
conv != dm_derive_conv_id(author, recipient)) goto invalid;
if (!author || !recipient || author == recipient || !seq || seq > INT64_MAX || !conv || conv > INT64_MAX) goto invalid;
uint8_t x[32], ed[32];
if (dm_node_pubkeys(&view, recipient, x, ed) != 0 ||
sc_ed25519_verify(ed, receipt, 72, receipt + 72) != SC_OK) goto invalid;
if (msg) {
uint8_t digest[32];
size_t actual = 0;
if (!dm_msg_len(msg, len, &actual) || actual != len || memcmp(receipt + 8, msg, 16) ||
struct dm_message_view message;
if (dm_message_parse(msg, len, &message) || message.size != len || memcmp(receipt + 8, msg, 16) ||
memcmp(receipt + 24, msg + 24, 8)) goto invalid;
if (message.mode == DM_MODE_GROUPS) {
uint8_t target_ed[32];
if (dm_message_target(&message, recipient, NULL, target_ed) < 0 || memcmp(target_ed, ed, 32)) goto invalid;
} else if (conv != dm_derive_conv_id(author, recipient)) goto invalid;
SHA256(msg, len, digest);
if (memcmp(digest, receipt + 40, 32)) goto invalid;
}
@@ -680,10 +683,11 @@ invalid:
int dm_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEIPT_SIZE]) {
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || dm_verify_receipt(inst, receipt, NULL, 0) != 0) return -1;
uint64_t author, conv, seq;
uint64_t author, conv, seq, recipient;
memcpy(&author, receipt + 24, 8);
memcpy(&conv, receipt + 8, 8);
memcpy(&seq, receipt + 16, 8);
memcpy(&recipient, receipt + 32, 8);
if (author != inst->node_id) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: receipt for another sender=%llu", DM_ID, (unsigned long long)author);
return -1;
@@ -691,6 +695,7 @@ int dm_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEI
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv);
sqlite3_stmt* st = NULL;
int transaction = 0;
int rc = sqlite3_prepare_v2(dm->db, "SELECT body FROM dm_outbox WHERE conv_id=? AND seq=?", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
@@ -704,19 +709,37 @@ int dm_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEI
}
sqlite3_finalize(st);
st = NULL;
rc = sqlite3_prepare_v2(dm->db, "DELETE FROM dm_outbox WHERE conv_id=? AND seq=?", -1, &st, NULL);
if (dm_exec(dm, "BEGIN IMMEDIATE")) return -1;
transaction = 1;
rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_delivery SET receipt=? WHERE conv_id=? AND author=? AND seq=? AND recipient=?", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_blob(st, 1, receipt, DM_RECEIPT_SIZE, SQLITE_STATIC);
sqlite3_bind_text(st, 2, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)author); sqlite3_bind_int64(st, 4, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 5, (sqlite3_int64)recipient);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE || sqlite3_changes(dm->db) != 1) goto fail;
sqlite3_finalize(st); st = NULL;
rc = sqlite3_prepare_v2(dm->db, "DELETE FROM dm_outbox WHERE conv_id=? AND seq=? AND NOT EXISTS"
" (SELECT 1 FROM dm_delivery d WHERE d.conv_id=dm_outbox.conv_id AND d.seq=dm_outbox.seq AND d.author=? AND d.receipt IS NULL)",
-1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 3, (sqlite3_int64)author);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
sqlite3_finalize(st);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: delivery confirmed conv=%s seq=%llu", DM_ID, conv_id, (unsigned long long)seq);
int complete = sqlite3_changes(dm->db);
sqlite3_finalize(st); st = NULL;
if (dm_exec(dm, "COMMIT")) goto fail;
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: delivery confirmed conv=%s seq=%llu recipient=%llu complete=%d", DM_ID, conv_id,
(unsigned long long)seq, (unsigned long long)recipient, complete);
dm_media_notify(inst, conv);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receipt database error rc=%d sql=%s", DM_ID, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
if (transaction) dm_exec(dm, "ROLLBACK");
return -1;
}
@@ -726,8 +749,9 @@ static void dm_pump(struct dm_state* dm) {
for (int i = 0; i < 32; i++) {
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db,
"SELECT o.conv_id,o.seq,o.body,c.peer_node_id,c.group_id FROM dm_outbox o"
" JOIN dm_conversations c USING(conv_id) WHERE o.retry_at<=? ORDER BY o.retry_at,o.seq LIMIT 1", -1, &st, NULL);
"SELECT o.conv_id,o.seq,o.body,d.recipient,c.group_id,d.direct_until FROM dm_outbox o"
" JOIN dm_conversations c USING(conv_id) JOIN dm_delivery d ON d.conv_id=o.conv_id AND d.seq=o.seq"
" WHERE d.retry_at<=? AND d.receipt IS NULL AND c.closed=0 ORDER BY d.retry_at,o.seq,d.recipient LIMIT 1", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_int64(st, 1, now);
rc = sqlite3_step(st);
@@ -745,23 +769,32 @@ static void dm_pump(struct dm_state* dm) {
}
memcpy(body, sqlite3_column_blob(st, 2), len);
uint64_t peer = (uint64_t)sqlite3_column_int64(st, 3), gid = (uint64_t)sqlite3_column_int64(st, 4);
int64_t deadline = sqlite3_column_int64(st, 5);
if (!deadline) deadline = now + 5;
sqlite3_finalize(st);
st = NULL;
rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_outbox SET retry_at=? WHERE conv_id=? AND seq=?", -1, &st, NULL);
rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_delivery SET retry_at=?,direct_until=?,"
" first_attempt=CASE WHEN first_attempt=0 THEN ? ELSE first_attempt END WHERE conv_id=? AND seq=? AND recipient=?",
-1, &st, NULL);
if (rc == SQLITE_OK) {
sqlite3_bind_int64(st, 1, now + 5);
sqlite3_bind_text(st, 2, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 2, deadline); sqlite3_bind_int64(st, 3, now);
sqlite3_bind_text(st, 4, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 5, (sqlite3_int64)seq); sqlite3_bind_int64(st, 6, (sqlite3_int64)peer);
rc = sqlite3_step(st);
}
sqlite3_finalize(st);
st = NULL;
if (rc != SQLITE_DONE) { u_free(body); goto fail; }
if (dm_route_group(dm->inst, peer, gid, &gid)) {
int direct = dm_route_group(dm->inst, peer, gid, &gid);
if (direct) {
if (!dm_route_send(dm, gid, peer, DM_SUBCMD_MSG, body, len))
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: outbox direct conv=%s seq=%llu group=%llu", DM_ID,
conv, (unsigned long long)seq, (unsigned long long)gid);
} else {
}
if (!direct || now >= deadline) {
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM delivery: mailbox fallback conv=%s seq=%llu recipient=%llu reason=%s", conv,
(unsigned long long)seq, (unsigned long long)peer, direct ? "direct ACK deadline" : "no route");
dm_mailbox_put(dm->inst, peer, dm->inst->node_id, body, len);
}
u_free(body);
@@ -906,6 +939,39 @@ int dm_start(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
return 0;
}
/* Снимок доставки сохраняется только источником, в транзакции отправки. */
static int dm_queue_targets(struct dm_state* dm, const struct dm_conv* c, const uint8_t* body, size_t length) {
struct dm_message_view view;
if (dm_message_parse(body, length, &view) || view.size != length || view.author != dm->inst->node_id) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM queue: invalid source/body conv=%s", c->conv_id); return -1;
}
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db, "INSERT INTO dm_delivery(conv_id,author,seq,recipient) VALUES(?,?,?,?)", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
unsigned count = view.mode == DM_MODE_GROUPS ? view.target_count : 1, queued = 0;
for (unsigned i = 0; i < count; i++) {
struct dm_target target = {.node = c->peer_node_id, .side = 1};
if (view.mode == DM_MODE_GROUPS && dm_message_target_at(&view, i, &target)) goto fail;
if (target.node == dm->inst->node_id) continue;
rc = sqlite3_reset(st);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, c->conv_id, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)view.author);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)view.seq);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)target.node);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
queued++;
}
rc = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_OK || !queued) goto fail;
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM queue: frozen delivery list conv=%s seq=%llu recipients=%u", c->conv_id,
(unsigned long long)view.seq, queued);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM queue: recipient insert failed conv=%s rc=%d sql=%s", c->conv_id, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st); return -1;
}
/* Атомарно сохранить seq, историю и неизменяемое тело исходящей задачи. */
int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content_type,
const uint8_t* data, uint32_t data_len) {
@@ -919,10 +985,12 @@ int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: send conversation missing/seq exhausted conv=%s", DM_ID, conv_id);
return -1;
}
if (!dm_conv_active(dm, &c)) return -1;
if (!dm_history_current(inst, conv_id, dm_history_generation(inst, conv_id))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM send: hidden conversation conv=%s", conv_id); return -1;
}
uint64_t seq = c.last_out_seq + 1;
uint64_t seq;
if (dm_sequence_next(inst, &seq)) return -1;
uint64_t ts = (uint64_t)(ntp_time_get_us(inst) / 1000);
uint8_t* body = NULL;
size_t len = 0;
@@ -931,8 +999,8 @@ int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content
if (dm_exec(dm, "BEGIN IMMEDIATE") != 0) { u_free(body); return -1; }
c.last_out_seq = seq;
if (ts > c.last_ts) c.last_ts = ts;
if (dm_store_message(dm, conv_id, 1, body, len) != 1 || dm_conv_save(dm, &c) != 0 ||
dm_media_track(inst, 1, body, len) != 0) goto rollback;
if (dm_sequence_take(inst, seq) || dm_store_message(dm, conv_id, 1, body, len) != 1 || dm_conv_save(dm, &c) != 0 ||
dm_media_track(inst, 1, body, len) != 0 || dm_queue_targets(dm, &c, body, len)) goto rollback;
int rc = sqlite3_prepare_v2(dm->db, "INSERT INTO dm_outbox(conv_id,seq,body) VALUES(?,?,?)", -1, &st, NULL);
if (rc == SQLITE_OK) {
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);