Check atomic DM enqueue rollback and explain sequence conflicts

src/dm/dm_core.c

@@ -393,8 +393,13 @@ static int dm_store_message(struct dm_state* dm, const char* conv, int dir, cons
     sqlite3_bind_int(st, 2, dir);
     sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
     rc = sqlite3_step(st);
-    if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != DM_MSG_SIG_SIZE ||
-        memcmp(sqlite3_column_blob(st, 0), msg + len - DM_MSG_SIG_SIZE, DM_MSG_SIG_SIZE)) goto fail;
+    if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != DM_MSG_SIG_SIZE) goto fail;
+    if (memcmp(sqlite3_column_blob(st, 0), msg + len - DM_MSG_SIG_SIZE, DM_MSG_SIG_SIZE)) {
+        DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: conflicting body for saved message conv=%s dir=%d seq=%llu", DM_ID,
+                   conv, dir, (unsigned long long)seq);
+        sqlite3_finalize(st);
+        return -1;
+    }
     sqlite3_finalize(st);
     return 0;
 fail:

tests/test_dm_e2e.c

@@ -278,6 +278,19 @@ static int protocol_checks(struct UTUN_INSTANCE* A, struct UTUN_INSTANCE* B) {
         dm_accept_receipt(A, receipt) == 0 || outbox_count(A) != 1) return -1;
     receipt[DM_RECEIPT_SIZE - 1] ^= 1;
     if (dm_accept_receipt(A, receipt) == 0 || outbox_count(A) != 1) return -1;
+    /* Ошибка durable outbox откатывает одновременно историю и seq отправителя. */
+    if (sqlite3_exec(A->topo_sqlite_db, "CREATE TEMP TRIGGER dm_outbox_fail BEFORE INSERT ON dm_outbox"
+                     " BEGIN SELECT RAISE(ABORT,'test outbox failure'); END", NULL, NULL, NULL) != SQLITE_OK) return -1;
+    if (dm_send(A, g_sh.conv_id, "text", (const uint8_t*)"must-not-commit", 15) == 0 || outbox_count(A) != 1) return -1;
+    if (sqlite3_exec(A->topo_sqlite_db, "DROP TRIGGER dm_outbox_fail", NULL, NULL, NULL) != SQLITE_OK ||
+        dm_msg_has(A, "must-not-commit")) return -1;
+    sqlite3_stmt* st = NULL;
+    if (sqlite3_prepare_v2(A->topo_sqlite_db, "SELECT last_out_seq FROM dm_conversations WHERE conv_id=?", -1, &st, NULL) != SQLITE_OK)
+        return -1;
+    sqlite3_bind_text(st, 1, g_sh.conv_id, -1, SQLITE_STATIC);
+    int seq_unchanged = sqlite3_step(st) == SQLITE_ROW && sqlite3_column_int64(st, 0) == 1;
+    sqlite3_finalize(st);
+    if (!seq_unchanged) return -1;
     DEBUG_INFO(DEBUG_CATEGORY_DM, "protocol checks passed: reorder, dedup, tamper, DB rollback, exact receipt");
     return 0;
 }