From 9985b79aa63d2acbd46c3944325065d11adb05cc Mon Sep 17 00:00:00 2001 From: Julius Joska Date: Thu, 13 Aug 2026 12:23:24 +0200 Subject: [PATCH] fix(app): ping the relay, retry stuck messages, drop duplicates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A message sent into a dead socket ended up as failed — attachments even stayed on "sending" forever — and only the user could send it again by hand. Mobile NAT drops idle TCP silently, so neither side knew. - pingInterval on the WebSocket, so the client notices a dead link itself - outbox: after every reconnect, stuck messages are re-sent (batch of 30, up to a day old); permanent errors are not retried - sendAttachment gained the try/catch it was missing, otherwise the attachment never reached the outbox - de-duplicate by message id: a retry re-sends the SAME message, which would otherwise show up twice on the other side --- app/lib/core/network/relay_transport.dart | 14 +- app/lib/core/services/messaging_service.dart | 209 +++++++++++++++++-- app/lib/core/storage/secure_storage.dart | 8 +- app/lib/models/conversation.dart | 10 +- app/test/conversation_dedup_test.dart | 43 ++++ 5 files changed, 259 insertions(+), 25 deletions(-) create mode 100644 app/test/conversation_dedup_test.dart diff --git a/app/lib/core/network/relay_transport.dart b/app/lib/core/network/relay_transport.dart index abc3964..b21d983 100644 --- a/app/lib/core/network/relay_transport.dart +++ b/app/lib/core/network/relay_transport.dart @@ -21,6 +21,15 @@ import 'package:socks5_proxy/socks_client.dart'; import 'package:web_socket_channel/io.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; +/// Jak často posílat WS ping a jak dlouho čekat na pong. +/// +/// Bez pingu se mrtvé spojení tváří jako živé: mobilní NAT zahodí nečinné TCP +/// tiše, appka dál hlásí „připojeno" a píše do prázdna — zprávy pak odejdou až +/// po restartu appky, všechny naráz. `dart:io` při nastaveném intervalu ping +/// posílá a když pong nedorazí včas, spojení zavře → [RelayClient] se +/// reconnectne sám. Relay pinguje protisměrně po 30 s (ws.rs PING_INTERVAL). +const kWsPingInterval = Duration(seconds: 30); + /// Abstrakce nad tím, jak klient otevírá spojení k relayi. abstract class RelayTransport { /// Sdílený HTTP klient pro REST (registrace, prekey bundle, přílohy…). @@ -41,7 +50,8 @@ class DirectTransport implements RelayTransport { http.Client httpClient() => _client ??= http.Client(); @override - WebSocketChannel connectWs(Uri url) => WebSocketChannel.connect(url); + WebSocketChannel connectWs(Uri url) => + IOWebSocketChannel.connect(url, pingInterval: kWsPingInterval); @override void close() { @@ -84,7 +94,7 @@ class TorTransport implements RelayTransport { @override WebSocketChannel connectWs(Uri url) => IOWebSocketChannel.connect(_withExplicitPort(url), - customClient: _sharedHttpClient()); + customClient: _sharedHttpClient(), pingInterval: kWsPingInterval); /// `Uri` pro `ws`/`wss` nezná výchozí port → `url.port` vrací 0 a SOCKS /// CONNECT by cílil na port 0 (serverError). Doplníme ho podle schématu diff --git a/app/lib/core/services/messaging_service.dart b/app/lib/core/services/messaging_service.dart index 0b9464a..32ab505 100644 --- a/app/lib/core/services/messaging_service.dart +++ b/app/lib/core/services/messaging_service.dart @@ -78,9 +78,15 @@ class MessagingService { bool _trustRootMismatch = false; bool get trustRootMismatch => _trustRootMismatch; StreamSubscription? _deliverSub; + StreamSubscription? _stateSub; Timer? _registerRetryTimer; Timer? _purgeTimer; + /// Běží právě dodatečné odeslání zaseklých zpráv? Reconnectů může přijít + /// několik rychle za sebou a dva souběžné průchody by tutéž zprávu poslaly + /// dvakrát. + bool _outboxRunning = false; + /// Transport relaye: false = přímé WSS, true = přes Tor (SOCKS5/Orbot na /// `.onion`). Čte se z nastavení při startu, přepíná se za běhu /// ([setTorEnabled]) přestavením [RelayClient]. @@ -155,6 +161,10 @@ class MessagingService { final relay = _buildRelay(); _relay = relay; _deliverSub = relay.deliveries.listen(_onDeliver); + // Po každém (re)connectu dorazit, co zůstalo viset ve frontě odchozích. + _stateSub = relay.stateStream.listen((state) { + if (state == RelayConnectionState.connected) unawaited(_flushOutbox()); + }); if (existingAccount != null) { relay.restoreSession( @@ -211,6 +221,7 @@ class MessagingService { /// push registrace se znovu odešle. Trust root zůstává pinovaný. Future _rebuildRelay() async { await _deliverSub?.cancel(); + await _stateSub?.cancel(); await _relay?.dispose(); _senderCert = null; _certFetchedAt = null; @@ -218,6 +229,10 @@ class MessagingService { final relay = _buildRelay(); _relay = relay; _deliverSub = relay.deliveries.listen(_onDeliver); + // Po každém (re)connectu dorazit, co zůstalo viset ve frontě odchozích. + _stateSub = relay.stateStream.listen((state) { + if (state == RelayConnectionState.connected) unawaited(_flushOutbox()); + }); final existingAccount = accountId; if (existingAccount != null) { @@ -357,6 +372,7 @@ class MessagingService { 'v': 1, 't': 'txt', 'id': message.id, + 'ts': message.timestamp.millisecondsSinceEpoch, 'txt': text, 're': ?replyToId, 'ttl': ?ttl, @@ -409,23 +425,35 @@ class MessagingService { await storage.addMessage(contact.id, message); onUpdate?.call(); - final delivered = await _sendContent(contact, { - 'v': 1, - 't': 'att', - 'id': message.id, - if (caption != null && caption.isNotEmpty) 'txt': caption, - 'ttl': ?ttl, - 'att': meta.toWire(), - 'p': _profile(), - }); + // Stejné ošetření jako u textu: bez něj zůstala příloha po výpadku + // spojení viset na „odesílá se" navždy. Označit jako failed ji dostane + // do fronty odchozích ([_flushOutbox]) — blob je na relayi nahraný, + // dosílá se jen meta. + try { + final delivered = await _sendContent(contact, { + 'v': 1, + 't': 'att', + 'id': message.id, + 'ts': message.timestamp.millisecondsSinceEpoch, + if (caption != null && caption.isNotEmpty) 'txt': caption, + 'ttl': ?ttl, + 'att': meta.toWire(), + 'p': _profile(), + }); - final sent = message.copyWith( - isDelivered: delivered, - status: delivered ? MessageStatus.delivered : MessageStatus.sent, - ); - await storage.updateMessage(contact.id, sent); - onUpdate?.call(); - return sent; + final sent = message.copyWith( + isDelivered: delivered, + status: delivered ? MessageStatus.delivered : MessageStatus.sent, + ); + await storage.updateMessage(contact.id, sent); + onUpdate?.call(); + return sent; + } catch (_) { + await storage.updateMessage( + contact.id, message.copyWith(status: MessageStatus.failed)); + onUpdate?.call(); + rethrow; + } } /// Zpracuje a nahraje přílohu: obrázky re-enkóduje (EXIF/GPS pryč, @@ -703,6 +731,7 @@ class MessagingService { 'v': 1, 't': 'txt', 'id': message.id, + 'ts': message.timestamp.millisecondsSinceEpoch, 'gid': group.id, 'txt': text, 're': ?replyToId, @@ -759,6 +788,7 @@ class MessagingService { 'v': 1, 't': 'att', 'id': message.id, + 'ts': message.timestamp.millisecondsSinceEpoch, 'gid': group.id, if (caption != null && caption.isNotEmpty) 'txt': caption, 'ttl': ?ttl, @@ -1022,6 +1052,115 @@ class MessagingService { /// Společná odesílací cesta: session bootstrap (X3DH s anti-MITM /// kontrolou identity klíče), sealed encrypt, obálka na relay. /// Vrací true = doručeno online, false = uloženo do offline fronty. + /// Kolik zaseklých zpráv poslat na jeden průchod. Po delším výpadku sítě + /// jich může být hodně; dávka drží start spojení svižný a zbytek dojde + /// při dalším ticku. + static const _outboxBatch = 30; + + /// Jak staré zaseklé zprávy ještě dosílat. Co leží dýl, patří uživateli + /// do rukou — poslat týden starou zprávu bez zeptání by překvapilo. + static const _outboxMaxAge = Duration(hours: 24); + + /// Došle zprávy, které zůstaly viset po výpadku spojení. + /// + /// Bez tohohle skončila každá zpráva odeslaná do mrtvého socketu jako + /// `failed` (nebo věčné `sending` u příloh) a znovu ji poslal jedině + /// uživatel ručně. Volá se po každém úspěšném (re)connectu; id zprávy + /// zůstává stejné, takže protistrana duplicitu zahodí + /// ([Conversation.addMessage]). + Future _flushOutbox() async { + if (_outboxRunning || !isStarted) return; + _outboxRunning = true; + try { + final identity = _identity; + if (identity == null) return; + final cutoff = DateTime.now().subtract(_outboxMaxAge); + + final stuck = <(String convId, Message message)>[]; + for (final conv in await storage.loadConversations()) { + for (final m in conv.messages) { + if (m.senderId != identity.id) continue; + if (m.status != MessageStatus.failed && + m.status != MessageStatus.sending) { + continue; + } + if (m.timestamp.isBefore(cutoff) || m.isDeleted) continue; + stuck.add((conv.contactId, m)); + } + } + if (stuck.isEmpty) return; + + stuck.sort((a, b) => a.$2.timestamp.compareTo(b.$2.timestamp)); + for (final (convId, message) in stuck.take(_outboxBatch)) { + if (connectionState != RelayConnectionState.connected) break; + await _resend(convId, message); + } + onUpdate?.call(); + } finally { + _outboxRunning = false; + } + } + + /// Jeden pokus o dodatečné odeslání. Trvalé chyby (chybný klíč protistrany, + /// kontakt bez doručovací adresy) nechají zprávu `failed` a dál se + /// neopakují — opakovat má smysl jen to, co shodila síť. + Future _resend(String convId, Message message) async { + try { + final group = await storage.loadGroup(convId); + final bool delivered; + if (group != null) { + delivered = await _fanoutToGroup(group, _outboxWire(message, gid: convId)); + } else { + final contact = (await storage.loadContacts()) + .where((c) => c.id == convId) + .firstOrNull; + if (contact == null) return; + delivered = await _sendContent(contact, _outboxWire(message)); + } + await storage.updateMessage( + convId, + message.copyWith( + isDelivered: delivered, + status: delivered ? MessageStatus.delivered : MessageStatus.sent, + ), + ); + } on SepotUserError { + // Srozumitelná chyba = stav, který se opakováním nespraví (chybí + // doručovací adresa, nesedí identity klíč, podvržený relay). + await storage.updateMessage( + convId, + message.copyWith(status: MessageStatus.failed), + ); + } on Object { + // Síť/timeout — nechat viset a zkusit po příštím reconnectu. + await storage.updateMessage( + convId, + message.copyWith(status: MessageStatus.failed), + ); + } + } + + /// Obsah zprávy pro drát při dodatečném odeslání. Drží se tvaru, který + /// posílá [sendText]/[sendAttachment] — včetně `id`, aby šlo o TUTÉŽ + /// zprávu, ne novou. + Map _outboxWire(Message message, {String? gid}) { + final isAttachment = message.attachment != null; + final caption = message.content; + return { + 'v': 1, + 't': isAttachment ? 'att' : 'txt', + 'id': message.id, + 'ts': message.timestamp.millisecondsSinceEpoch, + 'gid': ?gid, + if (isAttachment) 'att': message.attachment!.toWire(), + if (!isAttachment || (caption != null && caption.isNotEmpty)) + 'txt': caption ?? '', + 're': ?message.replyToId, + 'ttl': ?message.disappearAfterSeconds, + 'p': _profile(), + }; + } + Future _sendContent( Contact contact, Map content, @@ -1231,7 +1370,7 @@ class MessagingService { type: MessageType.text, content: text, replyToId: replyTo, - timestamp: DateTime.now(), + timestamp: wireSentAt(content['ts']), isDelivered: true, disappearAfterSeconds: ttl, ); @@ -1257,18 +1396,31 @@ class MessagingService { fileName: meta.name, fileSize: meta.size, replyToId: replyTo, - timestamp: DateTime.now(), + timestamp: wireSentAt(content['ts']), isDelivered: true, attachment: meta, disappearAfterSeconds: ttl, ); } - await storage.addMessage(convId, message); + final isNew = await storage.addMessage(convId, message); // Uloženo na disku → teprve teď smí relay obálku zahodit. // (Blob přílohy do fronty nepatří — stahuje se na pozadí a při // neúspěchu má UI retry, ACK na něj nečeká.) _relay!.ack(envelope.msgUuid); + + // Zprávu už známe (odesílatel ji poslal znovu po výpadku spojení). + // Druhá bublina ani další stahování téhož blobu nemá vzniknout, ale + // doručenku pošli — protistrana ji nejspíš právě proto nedostala. + if (!isNew) { + unawaited(_sendControl(contact, { + 't': 'dlv', + 'ref': message.id, + 'gid': ?gid, + })); + return; + } + onUpdate?.call(); onMessageReceived?.call(contact, message); @@ -1300,6 +1452,24 @@ class MessagingService { } } + /// Čas odeslání z drátu (ms). Bez něj by zpráva dostala čas doručení — + /// po vyzvednutí offline fronty by celá konverzace uskočila na „teď" + /// a každý člen skupiny by u téže zprávy viděl jiný čas. + /// + /// Klienti ≤1.1.1 `ts` neposílají → fallback na čas doručení (chová se + /// jako dřív). Odesílatel si čas diktuje sám (jde E2E kanálem), proto + /// strop: hodiny zařízení se běžně rozcházejí o minuty, ale zpráva + /// nesmí uskočit mimo konverzaci. + @visibleForTesting + static DateTime wireSentAt(Object? raw) { + final now = DateTime.now(); + if (raw is! int) return now; + final sent = DateTime.fromMillisecondsSinceEpoch(raw); + if (sent.isAfter(now.add(const Duration(minutes: 5)))) return now; + if (sent.isBefore(DateTime(2020))) return now; + return sent; + } + /// Aplikuje příchozí reakci/edit/delete na cílovou zprávu v dané /// konverzaci (1:1 = kontakt, skupina = gid). Klíč reakce = fingerprint /// odesílatele (kryptograficky ověřený sealed senderem, ne @@ -1514,6 +1684,7 @@ class MessagingService { _purgeTimer?.cancel(); _purgeTimer = null; await _deliverSub?.cancel(); + await _stateSub?.cancel(); await _relay?.dispose(); // DB vlastní SecureStorageService (lock/deleteAll) — tady jen pustit _db = null; diff --git a/app/lib/core/storage/secure_storage.dart b/app/lib/core/storage/secure_storage.dart index 9b5bb92..5ee5bf9 100644 --- a/app/lib/core/storage/secure_storage.dart +++ b/app/lib/core/storage/secure_storage.dart @@ -442,11 +442,15 @@ class SecureStorageService { ); } - Future addMessage(String contactId, Message message) async { + /// Přidá zprávu do konverzace. Vrací false, když už tam zpráva s tímhle id + /// je (opakované doručení) — volající pak přeskočí navazující kroky jako + /// doručenku nebo stahování přílohy. + Future addMessage(String contactId, Message message) async { var conv = await loadConversation(contactId); conv ??= Conversation(contactId: contactId); - conv.addMessage(message); + if (!conv.addMessage(message)) return false; await saveConversation(conv); + return true; } /// Založí prázdnou konverzaci, pokud ještě není (nová skupina musí být diff --git a/app/lib/models/conversation.dart b/app/lib/models/conversation.dart index 097b41b..f70e360 100644 --- a/app/lib/models/conversation.dart +++ b/app/lib/models/conversation.dart @@ -39,10 +39,16 @@ class Conversation { .length; } - /// Přidej zprávu - void addMessage(Message message) { + /// Přidej zprávu. Vrací false, když už zpráva s tímhle id v konverzaci je. + /// + /// Id zprávy nese odesílatel (pole `id` na drátě), takže opakované doručení + /// téhož obsahu — retry odesílatele po výpadku spojení nebo redelivery + /// z fronty relaye — nesmí založit druhou bublinu. + bool addMessage(Message message) { + if (messages.any((m) => m.id == message.id)) return false; messages.add(message); lastActivity = DateTime.now(); + return true; } /// Nahraď zprávu se stejným id (změna stavu přílohy, doručenka…). diff --git a/app/test/conversation_dedup_test.dart b/app/test/conversation_dedup_test.dart new file mode 100644 index 0000000..7908373 --- /dev/null +++ b/app/test/conversation_dedup_test.dart @@ -0,0 +1,43 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:sepot/models/conversation.dart'; +import 'package:sepot/models/message.dart'; + +/// Id zprávy nese odesílatel, takže tatáž zpráva může dorazit dvakrát: +/// odesílatel ji po výpadku spojení pošle znovu (fronta odchozích) nebo ji +/// relay doručí opakovaně, než dorazí ACK. Ani v jednom případě nesmí +/// vzniknout druhá bublina. +void main() { + Message zprava(String id, {String text = 'ahoj'}) => Message( + id: id, + senderId: 'alice', + recipientId: 'bob', + type: MessageType.text, + content: text, + timestamp: DateTime.parse('2026-08-13T09:00:00.000'), + ); + + test('zpráva se stejným id se podruhé nepřidá', () { + final c = Conversation(contactId: 'bob'); + + expect(c.addMessage(zprava('m1')), isTrue, reason: 'první doručení'); + expect(c.addMessage(zprava('m1')), isFalse, reason: 'opakované doručení'); + expect(c.messages.length, 1); + }); + + test('dedup se řídí id, ne obsahem — stejný text jinou zprávou projde', () { + final c = Conversation(contactId: 'bob'); + + expect(c.addMessage(zprava('m1', text: 'ahoj')), isTrue); + expect(c.addMessage(zprava('m2', text: 'ahoj')), isTrue); + expect(c.messages.length, 2); + }); + + test('duplicita nechá původní zprávu být', () { + final c = Conversation(contactId: 'bob'); + c.addMessage(zprava('m1', text: 'původní')); + + c.addMessage(zprava('m1', text: 'přepsaný')); + + expect(c.messages.single.content, 'původní'); + }); +}