From 1b87f3efe90ca51b79d07dfefb5932fa20ae183b Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 12:36:14 +0100 Subject: [PATCH] Deliver immutable messages per device with ACK deadline fallback --- src/dm/dm_core.c | 104 +++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 86 insertions(+), 18 deletions(-) diff --git a/src/dm/dm_core.c b/src/dm/dm_core.c index 88a76cadd..2678489e6 100644 --- a/src/dm/dm_core.c +++ b/src/dm/dm_core.c @@ -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);