From 8c2e5687754735cd7534e66063cf2e400d6e4a12 Mon Sep 17 00:00:00 2001 From: otsmr Date: Mon, 31 Aug 2026 21:31:55 +0200 Subject: [PATCH] remove unused database notification --- lib/src/database/rust_change_notifier.dart | 4 +- lib/src/services/api/api.service.dart | 10 ++ .../visual/components/app_outdated.comp.dart | 14 +- .../views/chats/chat_messages.view.dart | 15 +- .../animated_new_message.dart | 8 +- .../message_input.dart | 24 ++-- ...48b2e2510cc8ca43f9f418e60bfc5afb40d48.json | 21 --- ...41bb914b6f3a319967765240c1ac958768006.json | 12 -- ...86f164ec766a30c36bcc3984164005786e095.json | 12 ++ ...27f8e9e21940fb6d089fb1b8871d756577db3.json | 12 ++ ...828524a4450590d72e2310eb89bb7033f4399.json | 12 -- ...28a2e7637c422eb966d0a0b89b97ec7a9a8fb.json | 12 ++ ...336652a52d58403ce2e2552b53de6cb8b8743.json | 21 --- ...bb5f158b7fadb9b74db4fb922e26af0443a63.json | 26 ---- ...9e74d681510c978d27da3b6a132ca7e402c7a.json | 12 ++ ...a9dd1913ecf014c117f3b7d77e6d8267dd7d6.json | 12 -- ...fc4df47a128d2466e7fa9f224c1365501dd0b.json | 26 ++++ ...79522c5628f5b5e61a45904283734e14d645a.json | 12 -- ...ac482d6f7165af2fa5a2e275facb310acdaee.json | 12 -- ...843bc98cc0adb75add13334576af560cbfb64.json | 12 ++ ...779b5fe4f003a16ba01ee2834f028fabf213f.json | 12 -- ...1ec547556fe464ceb3993d595f8f2c7d8532f.json | 21 --- ...84c94407918776b0ad2b8c47a1bfb041b620b.json | 12 -- ...bfd69afb18384d91764009abb50610c959374.json | 21 --- ...5b060d389a1a07f54757b0e12ac4d16cc2ec9.json | 12 -- ...11a582bd664e168ed16e0ff6adb7532bddaec.json | 12 -- ...d5f8b5f6adc5fec457e6a32a6be51b4b0d122.json | 12 ++ ...d72ae5c0b8d1176a6c9860716832c1d82885b.json | 26 ---- ...8d103efaa97f89c7a3fa9d5ed4c8292a246a4.json | 38 ++++++ rust/build.rs | 22 +-- rust/src/api/messages/incoming.rs | 112 +++++++++------ rust/src/api/messages/incoming/messages.rs | 4 - rust/src/api/messages/outgoing.rs | 1 - rust/src/api/runtime.rs | 1 - rust/src/api/runtime/auth.rs | 63 ++++++++- rust/src/api/runtime/client.rs | 128 +++++++++++++++++- rust/src/api/runtime/request.rs | 83 ++++++++---- rust/src/api/server/contacts.rs | 1 - rust/src/bridge/wrapper/user_discovery.rs | 1 - rust/src/database/app/legacy_import.rs | 1 - .../app/migrations/0008_pending_plaintext.sql | 12 ++ rust/src/database/app/mod.rs | 24 ++-- rust/src/database/app/tables/receipt.rs | 120 ++++++++++++++++ rust/src/error.rs | 7 + rust/src/log.rs | 121 ++++++++++++++++- rust/src/native/video.rs | 1 - rust/src/services/contacts.rs | 4 - rust/src/services/direct_media_upload.rs | 1 - rust/src/services/groups/mod.rs | 7 - rust/src/services/media_upload.rs | 14 -- rust/src/services/mediafiles.rs | 5 - rust/src/services/messages.rs | 69 ++++++---- rust/src/services/notifications.rs | 10 +- rust/src/user_discovery.rs | 8 +- rust/tests/api/tester.rs | 2 - 55 files changed, 842 insertions(+), 435 deletions(-) delete mode 100644 rust/.sqlx/query-0e2dcd586093544fb7c0163785648b2e2510cc8ca43f9f418e60bfc5afb40d48.json delete mode 100644 rust/.sqlx/query-1c0946be221d955a0082ba67f7041bb914b6f3a319967765240c1ac958768006.json create mode 100644 rust/.sqlx/query-1e99ff0bf394ff271be3173e86c86f164ec766a30c36bcc3984164005786e095.json create mode 100644 rust/.sqlx/query-231815a391f5a44481e92b316c927f8e9e21940fb6d089fb1b8871d756577db3.json delete mode 100644 rust/.sqlx/query-30e32daf98cb4d71d84b79f214d828524a4450590d72e2310eb89bb7033f4399.json create mode 100644 rust/.sqlx/query-36d08507990dc90860bde55e44e28a2e7637c422eb966d0a0b89b97ec7a9a8fb.json delete mode 100644 rust/.sqlx/query-3ef50101727d2255e0ae23e4151336652a52d58403ce2e2552b53de6cb8b8743.json delete mode 100644 rust/.sqlx/query-407232f4e5dae0ed496d80369d0bb5f158b7fadb9b74db4fb922e26af0443a63.json create mode 100644 rust/.sqlx/query-4bd3233fbadc3e75f3b929571fa9e74d681510c978d27da3b6a132ca7e402c7a.json delete mode 100644 rust/.sqlx/query-4d18233931fbadbfc01254449b9a9dd1913ecf014c117f3b7d77e6d8267dd7d6.json create mode 100644 rust/.sqlx/query-6eef41e5ed8b8d8b3c390baadc3fc4df47a128d2466e7fa9f224c1365501dd0b.json delete mode 100644 rust/.sqlx/query-738756d95a2482031a39318684279522c5628f5b5e61a45904283734e14d645a.json delete mode 100644 rust/.sqlx/query-74c9a8f699b6a1c214e196824b9ac482d6f7165af2fa5a2e275facb310acdaee.json create mode 100644 rust/.sqlx/query-b579b1074dec0274fa72658bcb0843bc98cc0adb75add13334576af560cbfb64.json delete mode 100644 rust/.sqlx/query-c91dc6e6cff692209fa18216e2b1ec547556fe464ceb3993d595f8f2c7d8532f.json delete mode 100644 rust/.sqlx/query-d36767ac50047037ce2122bd59b84c94407918776b0ad2b8c47a1bfb041b620b.json delete mode 100644 rust/.sqlx/query-d84d22b95288a8fd4bbcb96bdddbfd69afb18384d91764009abb50610c959374.json delete mode 100644 rust/.sqlx/query-e24b46f465ca104c6f288eafda55b060d389a1a07f54757b0e12ac4d16cc2ec9.json delete mode 100644 rust/.sqlx/query-e33bd4d14da2ab7a0b083d894f911a582bd664e168ed16e0ff6adb7532bddaec.json create mode 100644 rust/.sqlx/query-e9d9eb36852d29f511a54e707d3d5f8b5f6adc5fec457e6a32a6be51b4b0d122.json delete mode 100644 rust/.sqlx/query-f4af415f434d17c2880be292539d72ae5c0b8d1176a6c9860716832c1d82885b.json create mode 100644 rust/.sqlx/query-fbc5fa19cad3ddde6098197c6d58d103efaa97f89c7a3fa9d5ed4c8292a246a4.json create mode 100644 rust/src/database/app/migrations/0008_pending_plaintext.sql diff --git a/lib/src/database/rust_change_notifier.dart b/lib/src/database/rust_change_notifier.dart index 28705ffb..9aa4792d 100644 --- a/lib/src/database/rust_change_notifier.dart +++ b/lib/src/database/rust_change_notifier.dart @@ -12,8 +12,8 @@ import 'package:twonly/src/utils/log.dart'; /// learns about it, so `watch()` keeps serving stale rows and the UI does not /// update until something else happens to touch the same table. /// -/// Rust already broadcasts the tables it commits to (`notify_committed`); this -/// forwards those batches into [GeneratedDatabase.notifyUpdates]. +/// SQLite's own commit hook broadcasts the tables each committed transaction +/// touched; this forwards those batches into [GeneratedDatabase.notifyUpdates]. StreamSubscription> listenToRustDatabaseChanges( GeneratedDatabase db, ) { diff --git a/lib/src/services/api/api.service.dart b/lib/src/services/api/api.service.dart index 90a3a1bb..e9b1e076 100644 --- a/lib/src/services/api/api.service.dart +++ b/lib/src/services/api/api.service.dart @@ -29,10 +29,20 @@ class ApiService { StreamSubscription>? _connectivitySubscription; late final StreamSubscription _apiEventSubscription; + /// The server rejects an outdated app or a superseded device during the + /// handshake, which usually happens before the widget showing that banner is + /// mounted. [events] is a broadcast stream, so late listeners would miss it — + /// they read the last rejection from here instead. + ApiEventKind? permanentRejection; + Future _handleApiEvent(ApiEvent event) async { if (event.kind == ApiEventKind.authenticated) { await onAuthenticated(); } + if (event.kind == ApiEventKind.appOutdated || + event.kind == ApiEventKind.newDeviceRegistered) { + permanentRejection = event.kind; + } } // Function is called after the user is authenticated at the server diff --git a/lib/src/visual/components/app_outdated.comp.dart b/lib/src/visual/components/app_outdated.comp.dart index d913eb40..52895192 100644 --- a/lib/src/visual/components/app_outdated.comp.dart +++ b/lib/src/visual/components/app_outdated.comp.dart @@ -30,19 +30,25 @@ class _AppOutdatedCompState extends State { @override void initState() { super.initState(); + // The rejection is sent during the API handshake, which regularly completes + // before this widget is mounted, so start from the last one the service saw. + _showRejection(apiService.permanentRejection); _apiEventSubscription = apiService.events.listen((event) async { if (!mounted) return; if (event.kind == ApiEventKind.appOutdated || event.kind == ApiEventKind.newDeviceRegistered) { await context.read().updateConnectionState(false); - setState(() { - appIsOutdated = event.kind == ApiEventKind.appOutdated; - newDeviceRegistered = event.kind == ApiEventKind.newDeviceRegistered; - }); + if (!mounted) return; + setState(() => _showRejection(event.kind)); } }); } + void _showRejection(ApiEventKind? kind) { + appIsOutdated = kind == ApiEventKind.appOutdated; + newDeviceRegistered = kind == ApiEventKind.newDeviceRegistered; + } + @override Widget build(BuildContext context) { if (newDeviceRegistered) { diff --git a/lib/src/visual/views/chats/chat_messages.view.dart b/lib/src/visual/views/chats/chat_messages.view.dart index 4335679e..ddda6a6a 100644 --- a/lib/src/visual/views/chats/chat_messages.view.dart +++ b/lib/src/visual/views/chats/chat_messages.view.dart @@ -35,6 +35,7 @@ import 'package:twonly/src/visual/views/chats/chat_messages_components/typing_in class _MessageAnimationState { bool hasReceivedFirstBatch = false; + DateTime? newestKnownAt; final HashSet knownMessageIds = HashSet(); final HashSet animateMessageIds = HashSet(); final HashSet reportedOpenedMessageIds = HashSet(); @@ -379,13 +380,22 @@ class _ChatMessagesViewState extends State List groupActions, { bool reportOpened = false, }) async { + final newestKnownAt = _animationState.newestKnownAt; for (final msg in newMessages) { + // Only messages appended after the newest one already loaded are new to + // the user. Messages fetched by scrolling up are older and must not + // animate, no matter who sent them. if (_animationState.hasReceivedFirstBatch && - !_animationState.knownMessageIds.contains(msg.messageId) && - msg.senderId == null) { + newestKnownAt != null && + msg.createdAt.isAfter(newestKnownAt) && + !_animationState.knownMessageIds.contains(msg.messageId)) { _animationState.animateMessageIds.add(msg.messageId); } _animationState.knownMessageIds.add(msg.messageId); + final currentNewest = _animationState.newestKnownAt; + if (currentNewest == null || msg.createdAt.isAfter(currentNewest)) { + _animationState.newestKnownAt = msg.createdAt; + } } final chatItems = []; @@ -680,6 +690,7 @@ class _ChatMessagesViewState extends State key: Key('anim_${chatMessage.messageId}'), messageId: chatMessage.messageId, animateIds: _animationState.animateMessageIds, + isOwnMessage: chatMessage.senderId == null, child: ChatListEntry( key: Key(chatMessage.messageId), message: _data.chatItems[i].message!, diff --git a/lib/src/visual/views/chats/chat_messages_components/animated_new_message.dart b/lib/src/visual/views/chats/chat_messages_components/animated_new_message.dart index 35e92966..27935eba 100644 --- a/lib/src/visual/views/chats/chat_messages_components/animated_new_message.dart +++ b/lib/src/visual/views/chats/chat_messages_components/animated_new_message.dart @@ -5,6 +5,7 @@ class AnimatedNewMessage extends StatefulWidget { required this.child, required this.messageId, required this.animateIds, + required this.isOwnMessage, super.key, }); @@ -12,6 +13,9 @@ class AnimatedNewMessage extends StatefulWidget { final String messageId; final Set animateIds; + /// Own messages grow out of the right edge, received ones out of the left. + final bool isOwnMessage; + @override State createState() => _AnimatedNewMessageState(); } @@ -79,7 +83,9 @@ class _AnimatedNewMessageState extends State alignment: Alignment.bottomLeft, child: ScaleTransition( scale: _scaleAnimation, - alignment: Alignment.bottomRight, + alignment: widget.isOwnMessage + ? Alignment.bottomRight + : Alignment.bottomLeft, child: FadeTransition( opacity: _opacityAnimation, child: widget.child, diff --git a/lib/src/visual/views/chats/chat_messages_components/message_input.dart b/lib/src/visual/views/chats/chat_messages_components/message_input.dart index 6a000553..22881ccc 100644 --- a/lib/src/visual/views/chats/chat_messages_components/message_input.dart +++ b/lib/src/visual/views/chats/chat_messages_components/message_input.dart @@ -71,18 +71,26 @@ class _MessageInputState extends State { Timer? _recordingTimer; DateTime? _recordingStartTime; - Future _sendMessage() async { - if (_textFieldController.text == '') return; - - await RustApi.insertAndSendText( - groupId: widget.group.groupId, - text: _textFieldController.text, - quoteMessageId: widget.quotesMessage?.messageId, - ); + void _sendMessage() { + final text = _textFieldController.text; + if (text == '') return; + final quoteMessageId = widget.quotesMessage?.messageId; + // Emptying the composer is not allowed to wait on the bridge: Rust commits + // the message row before it starts delivering, so the bubble is already on + // its way into the chat list while this call is still running. _textFieldController.clear(); widget.onMessageSend(); setState(() {}); + + unawaitedRustCall( + RustApi.insertAndSendText( + groupId: widget.group.groupId, + text: text, + quoteMessageId: quoteMessageId, + ), + 'insertAndSendText', + ); } @override diff --git a/rust/.sqlx/query-0e2dcd586093544fb7c0163785648b2e2510cc8ca43f9f418e60bfc5afb40d48.json b/rust/.sqlx/query-0e2dcd586093544fb7c0163785648b2e2510cc8ca43f9f418e60bfc5afb40d48.json deleted file mode 100644 index 43a607ab..00000000 --- a/rust/.sqlx/query-0e2dcd586093544fb7c0163785648b2e2510cc8ca43f9f418e60bfc5afb40d48.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "db_name": "SQLite", - "query": "SELECT EXISTS(SELECT 1 FROM contacts WHERE user_id = ? AND sealed_sender_enabled = 1)", - "describe": { - "columns": [ - { - "name": "EXISTS(SELECT 1 FROM contacts WHERE user_id = ? AND sealed_sender_enabled = 1)", - "ordinal": 0, - "type_info": "Integer", - "origin": "Expression" - } - ], - "parameters": { - "Right": 1 - }, - "nullable": [ - false - ] - }, - "hash": "0e2dcd586093544fb7c0163785648b2e2510cc8ca43f9f418e60bfc5afb40d48" -} diff --git a/rust/.sqlx/query-1c0946be221d955a0082ba67f7041bb914b6f3a319967765240c1ac958768006.json b/rust/.sqlx/query-1c0946be221d955a0082ba67f7041bb914b6f3a319967765240c1ac958768006.json deleted file mode 100644 index 1fcd79eb..00000000 --- a/rust/.sqlx/query-1c0946be221d955a0082ba67f7041bb914b6f3a319967765240c1ac958768006.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "DELETE FROM message_actions\n WHERE message_id = ? AND contact_id = ? AND type = 'sealedSenderAt'", - "describe": { - "columns": [], - "parameters": { - "Right": 2 - }, - "nullable": [] - }, - "hash": "1c0946be221d955a0082ba67f7041bb914b6f3a319967765240c1ac958768006" -} diff --git a/rust/.sqlx/query-1e99ff0bf394ff271be3173e86c86f164ec766a30c36bcc3984164005786e095.json b/rust/.sqlx/query-1e99ff0bf394ff271be3173e86c86f164ec766a30c36bcc3984164005786e095.json new file mode 100644 index 00000000..04626b60 --- /dev/null +++ b/rust/.sqlx/query-1e99ff0bf394ff271be3173e86c86f164ec766a30c36bcc3984164005786e095.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM user_discovery_shares WHERE 1", + "describe": { + "columns": [], + "parameters": { + "Right": 0 + }, + "nullable": [] + }, + "hash": "1e99ff0bf394ff271be3173e86c86f164ec766a30c36bcc3984164005786e095" +} diff --git a/rust/.sqlx/query-231815a391f5a44481e92b316c927f8e9e21940fb6d089fb1b8871d756577db3.json b/rust/.sqlx/query-231815a391f5a44481e92b316c927f8e9e21940fb6d089fb1b8871d756577db3.json new file mode 100644 index 00000000..87eef574 --- /dev/null +++ b/rust/.sqlx/query-231815a391f5a44481e92b316c927f8e9e21940fb6d089fb1b8871d756577db3.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n UPDATE received_receipts\n SET pending_plaintext = ?\n WHERE receipt_id = ?\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 2 + }, + "nullable": [] + }, + "hash": "231815a391f5a44481e92b316c927f8e9e21940fb6d089fb1b8871d756577db3" +} diff --git a/rust/.sqlx/query-30e32daf98cb4d71d84b79f214d828524a4450590d72e2310eb89bb7033f4399.json b/rust/.sqlx/query-30e32daf98cb4d71d84b79f214d828524a4450590d72e2310eb89bb7033f4399.json deleted file mode 100644 index 9cbc586e..00000000 --- a/rust/.sqlx/query-30e32daf98cb4d71d84b79f214d828524a4450590d72e2310eb89bb7033f4399.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "INSERT INTO privacy_pass_tokens(token, expires_at) VALUES (?, ?)\n ON CONFLICT(token) DO NOTHING", - "describe": { - "columns": [], - "parameters": { - "Right": 2 - }, - "nullable": [] - }, - "hash": "30e32daf98cb4d71d84b79f214d828524a4450590d72e2310eb89bb7033f4399" -} diff --git a/rust/.sqlx/query-36d08507990dc90860bde55e44e28a2e7637c422eb966d0a0b89b97ec7a9a8fb.json b/rust/.sqlx/query-36d08507990dc90860bde55e44e28a2e7637c422eb966d0a0b89b97ec7a9a8fb.json new file mode 100644 index 00000000..e4b61db4 --- /dev/null +++ b/rust/.sqlx/query-36d08507990dc90860bde55e44e28a2e7637c422eb966d0a0b89b97ec7a9a8fb.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO contacts(user_id, username, avatar_svg_compressed) VALUES (?, ?, ?)", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "36d08507990dc90860bde55e44e28a2e7637c422eb966d0a0b89b97ec7a9a8fb" +} diff --git a/rust/.sqlx/query-3ef50101727d2255e0ae23e4151336652a52d58403ce2e2552b53de6cb8b8743.json b/rust/.sqlx/query-3ef50101727d2255e0ae23e4151336652a52d58403ce2e2552b53de6cb8b8743.json deleted file mode 100644 index 8cf71a24..00000000 --- a/rust/.sqlx/query-3ef50101727d2255e0ae23e4151336652a52d58403ce2e2552b53de6cb8b8743.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "db_name": "SQLite", - "query": "SELECT COUNT(*) FROM privacy_pass_tokens WHERE expires_at > ?", - "describe": { - "columns": [ - { - "name": "COUNT(*)", - "ordinal": 0, - "type_info": "Integer", - "origin": "Expression" - } - ], - "parameters": { - "Right": 1 - }, - "nullable": [ - false - ] - }, - "hash": "3ef50101727d2255e0ae23e4151336652a52d58403ce2e2552b53de6cb8b8743" -} diff --git a/rust/.sqlx/query-407232f4e5dae0ed496d80369d0bb5f158b7fadb9b74db4fb922e26af0443a63.json b/rust/.sqlx/query-407232f4e5dae0ed496d80369d0bb5f158b7fadb9b74db4fb922e26af0443a63.json deleted file mode 100644 index 2883a0c8..00000000 --- a/rust/.sqlx/query-407232f4e5dae0ed496d80369d0bb5f158b7fadb9b74db4fb922e26af0443a63.json +++ /dev/null @@ -1,26 +0,0 @@ -{ - "db_name": "SQLite", - "query": "SELECT sealed_sender_enabled FROM contacts WHERE user_id = ?", - "describe": { - "columns": [ - { - "name": "sealed_sender_enabled", - "ordinal": 0, - "type_info": "Integer", - "origin": { - "Table": { - "table": "contacts", - "name": "sealed_sender_enabled" - } - } - } - ], - "parameters": { - "Right": 1 - }, - "nullable": [ - false - ] - }, - "hash": "407232f4e5dae0ed496d80369d0bb5f158b7fadb9b74db4fb922e26af0443a63" -} diff --git a/rust/.sqlx/query-4bd3233fbadc3e75f3b929571fa9e74d681510c978d27da3b6a132ca7e402c7a.json b/rust/.sqlx/query-4bd3233fbadc3e75f3b929571fa9e74d681510c978d27da3b6a132ca7e402c7a.json new file mode 100644 index 00000000..843e475f --- /dev/null +++ b/rust/.sqlx/query-4bd3233fbadc3e75f3b929571fa9e74d681510c978d27da3b6a132ca7e402c7a.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO contacts(user_id, username) VALUES (?, ?)", + "describe": { + "columns": [], + "parameters": { + "Right": 2 + }, + "nullable": [] + }, + "hash": "4bd3233fbadc3e75f3b929571fa9e74d681510c978d27da3b6a132ca7e402c7a" +} diff --git a/rust/.sqlx/query-4d18233931fbadbfc01254449b9a9dd1913ecf014c117f3b7d77e6d8267dd7d6.json b/rust/.sqlx/query-4d18233931fbadbfc01254449b9a9dd1913ecf014c117f3b7d77e6d8267dd7d6.json deleted file mode 100644 index 64461470..00000000 --- a/rust/.sqlx/query-4d18233931fbadbfc01254449b9a9dd1913ecf014c117f3b7d77e6d8267dd7d6.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "\n INSERT INTO signal_identities (name, identity_key, timestamp)\n VALUES (?, ?, 0)\n ", - "describe": { - "columns": [], - "parameters": { - "Right": 2 - }, - "nullable": [] - }, - "hash": "4d18233931fbadbfc01254449b9a9dd1913ecf014c117f3b7d77e6d8267dd7d6" -} diff --git a/rust/.sqlx/query-6eef41e5ed8b8d8b3c390baadc3fc4df47a128d2466e7fa9f224c1365501dd0b.json b/rust/.sqlx/query-6eef41e5ed8b8d8b3c390baadc3fc4df47a128d2466e7fa9f224c1365501dd0b.json new file mode 100644 index 00000000..dfcbac5c --- /dev/null +++ b/rust/.sqlx/query-6eef41e5ed8b8d8b3c390baadc3fc4df47a128d2466e7fa9f224c1365501dd0b.json @@ -0,0 +1,26 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT pending_plaintext\n FROM received_receipts\n WHERE receipt_id = ?\n ", + "describe": { + "columns": [ + { + "name": "pending_plaintext", + "ordinal": 0, + "type_info": "Blob", + "origin": { + "Table": { + "table": "received_receipts", + "name": "pending_plaintext" + } + } + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + true + ] + }, + "hash": "6eef41e5ed8b8d8b3c390baadc3fc4df47a128d2466e7fa9f224c1365501dd0b" +} diff --git a/rust/.sqlx/query-738756d95a2482031a39318684279522c5628f5b5e61a45904283734e14d645a.json b/rust/.sqlx/query-738756d95a2482031a39318684279522c5628f5b5e61a45904283734e14d645a.json deleted file mode 100644 index c1259020..00000000 --- a/rust/.sqlx/query-738756d95a2482031a39318684279522c5628f5b5e61a45904283734e14d645a.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "\n INSERT INTO message_actions(message_id, contact_id, type)\n VALUES (?, ?, 'sealedSenderAt')\n ON CONFLICT(message_id, contact_id, type)\n DO UPDATE SET action_at = CAST(strftime('%s', 'now') AS INTEGER)\n ", - "describe": { - "columns": [], - "parameters": { - "Right": 2 - }, - "nullable": [] - }, - "hash": "738756d95a2482031a39318684279522c5628f5b5e61a45904283734e14d645a" -} diff --git a/rust/.sqlx/query-74c9a8f699b6a1c214e196824b9ac482d6f7165af2fa5a2e275facb310acdaee.json b/rust/.sqlx/query-74c9a8f699b6a1c214e196824b9ac482d6f7165af2fa5a2e275facb310acdaee.json deleted file mode 100644 index f8c101a9..00000000 --- a/rust/.sqlx/query-74c9a8f699b6a1c214e196824b9ac482d6f7165af2fa5a2e275facb310acdaee.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "UPDATE contacts SET sealed_sender_enabled = ?\n WHERE user_id = ? AND sealed_sender_enabled != ?", - "describe": { - "columns": [], - "parameters": { - "Right": 3 - }, - "nullable": [] - }, - "hash": "74c9a8f699b6a1c214e196824b9ac482d6f7165af2fa5a2e275facb310acdaee" -} diff --git a/rust/.sqlx/query-b579b1074dec0274fa72658bcb0843bc98cc0adb75add13334576af560cbfb64.json b/rust/.sqlx/query-b579b1074dec0274fa72658bcb0843bc98cc0adb75add13334576af560cbfb64.json new file mode 100644 index 00000000..6403b092 --- /dev/null +++ b/rust/.sqlx/query-b579b1074dec0274fa72658bcb0843bc98cc0adb75add13334576af560cbfb64.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "UPDATE media_files SET download_state = 'ready', stored_file_hash = ?,\n size_in_bytes = ?\n WHERE media_id = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "b579b1074dec0274fa72658bcb0843bc98cc0adb75add13334576af560cbfb64" +} diff --git a/rust/.sqlx/query-c618a305e1355fcefa5c8420544779b5fe4f003a16ba01ee2834f028fabf213f.json b/rust/.sqlx/query-c618a305e1355fcefa5c8420544779b5fe4f003a16ba01ee2834f028fabf213f.json index 20d8c6d3..4f2971be 100644 --- a/rust/.sqlx/query-c618a305e1355fcefa5c8420544779b5fe4f003a16ba01ee2834f028fabf213f.json +++ b/rust/.sqlx/query-c618a305e1355fcefa5c8420544779b5fe4f003a16ba01ee2834f028fabf213f.json @@ -288,17 +288,6 @@ "name": "media_received_counter" } } - }, - { - "name": "sealed_sender_enabled", - "ordinal": 26, - "type_info": "Integer", - "origin": { - "Table": { - "table": "contacts", - "name": "sealed_sender_enabled" - } - } } ], "parameters": { @@ -330,7 +319,6 @@ true, true, false, - false, false ] }, diff --git a/rust/.sqlx/query-c91dc6e6cff692209fa18216e2b1ec547556fe464ceb3993d595f8f2c7d8532f.json b/rust/.sqlx/query-c91dc6e6cff692209fa18216e2b1ec547556fe464ceb3993d595f8f2c7d8532f.json deleted file mode 100644 index bf2b4d79..00000000 --- a/rust/.sqlx/query-c91dc6e6cff692209fa18216e2b1ec547556fe464ceb3993d595f8f2c7d8532f.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "db_name": "SQLite", - "query": "SELECT EXISTS(SELECT 1 FROM message_actions\n WHERE message_id = ? AND contact_id = ? AND type = 'sealedSenderAt')", - "describe": { - "columns": [ - { - "name": "EXISTS(SELECT 1 FROM message_actions\n WHERE message_id = ? AND contact_id = ? AND type = 'sealedSenderAt')", - "ordinal": 0, - "type_info": "Integer", - "origin": "Expression" - } - ], - "parameters": { - "Right": 2 - }, - "nullable": [ - false - ] - }, - "hash": "c91dc6e6cff692209fa18216e2b1ec547556fe464ceb3993d595f8f2c7d8532f" -} diff --git a/rust/.sqlx/query-d36767ac50047037ce2122bd59b84c94407918776b0ad2b8c47a1bfb041b620b.json b/rust/.sqlx/query-d36767ac50047037ce2122bd59b84c94407918776b0ad2b8c47a1bfb041b620b.json deleted file mode 100644 index 37dba35f..00000000 --- a/rust/.sqlx/query-d36767ac50047037ce2122bd59b84c94407918776b0ad2b8c47a1bfb041b620b.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "DELETE FROM user_discovery_shares", - "describe": { - "columns": [], - "parameters": { - "Right": 0 - }, - "nullable": [] - }, - "hash": "d36767ac50047037ce2122bd59b84c94407918776b0ad2b8c47a1bfb041b620b" -} diff --git a/rust/.sqlx/query-d84d22b95288a8fd4bbcb96bdddbfd69afb18384d91764009abb50610c959374.json b/rust/.sqlx/query-d84d22b95288a8fd4bbcb96bdddbfd69afb18384d91764009abb50610c959374.json deleted file mode 100644 index 8f59d827..00000000 --- a/rust/.sqlx/query-d84d22b95288a8fd4bbcb96bdddbfd69afb18384d91764009abb50610c959374.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "db_name": "SQLite", - "query": "SELECT COUNT(*) FROM privacy_pass_tokens", - "describe": { - "columns": [ - { - "name": "COUNT(*)", - "ordinal": 0, - "type_info": "Integer", - "origin": "Expression" - } - ], - "parameters": { - "Right": 0 - }, - "nullable": [ - false - ] - }, - "hash": "d84d22b95288a8fd4bbcb96bdddbfd69afb18384d91764009abb50610c959374" -} diff --git a/rust/.sqlx/query-e24b46f465ca104c6f288eafda55b060d389a1a07f54757b0e12ac4d16cc2ec9.json b/rust/.sqlx/query-e24b46f465ca104c6f288eafda55b060d389a1a07f54757b0e12ac4d16cc2ec9.json deleted file mode 100644 index 8c5766b4..00000000 --- a/rust/.sqlx/query-e24b46f465ca104c6f288eafda55b060d389a1a07f54757b0e12ac4d16cc2ec9.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "UPDATE media_files SET download_state = 'ready', stored_file_hash = ?\n WHERE media_id = ?", - "describe": { - "columns": [], - "parameters": { - "Right": 2 - }, - "nullable": [] - }, - "hash": "e24b46f465ca104c6f288eafda55b060d389a1a07f54757b0e12ac4d16cc2ec9" -} diff --git a/rust/.sqlx/query-e33bd4d14da2ab7a0b083d894f911a582bd664e168ed16e0ff6adb7532bddaec.json b/rust/.sqlx/query-e33bd4d14da2ab7a0b083d894f911a582bd664e168ed16e0ff6adb7532bddaec.json deleted file mode 100644 index 050669f6..00000000 --- a/rust/.sqlx/query-e33bd4d14da2ab7a0b083d894f911a582bd664e168ed16e0ff6adb7532bddaec.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "DELETE FROM privacy_pass_tokens WHERE expires_at <= ?", - "describe": { - "columns": [], - "parameters": { - "Right": 1 - }, - "nullable": [] - }, - "hash": "e33bd4d14da2ab7a0b083d894f911a582bd664e168ed16e0ff6adb7532bddaec" -} diff --git a/rust/.sqlx/query-e9d9eb36852d29f511a54e707d3d5f8b5f6adc5fec457e6a32a6be51b4b0d122.json b/rust/.sqlx/query-e9d9eb36852d29f511a54e707d3d5f8b5f6adc5fec457e6a32a6be51b4b0d122.json new file mode 100644 index 00000000..7d5e2306 --- /dev/null +++ b/rust/.sqlx/query-e9d9eb36852d29f511a54e707d3d5f8b5f6adc5fec457e6a32a6be51b4b0d122.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n UPDATE received_receipts\n SET pending_plaintext = NULL\n WHERE receipt_id = ? AND pending_plaintext IS NOT NULL\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "e9d9eb36852d29f511a54e707d3d5f8b5f6adc5fec457e6a32a6be51b4b0d122" +} diff --git a/rust/.sqlx/query-f4af415f434d17c2880be292539d72ae5c0b8d1176a6c9860716832c1d82885b.json b/rust/.sqlx/query-f4af415f434d17c2880be292539d72ae5c0b8d1176a6c9860716832c1d82885b.json deleted file mode 100644 index e560db61..00000000 --- a/rust/.sqlx/query-f4af415f434d17c2880be292539d72ae5c0b8d1176a6c9860716832c1d82885b.json +++ /dev/null @@ -1,26 +0,0 @@ -{ - "db_name": "SQLite", - "query": "DELETE FROM privacy_pass_tokens\n WHERE token = (\n SELECT token FROM privacy_pass_tokens\n WHERE expires_at > ?\n ORDER BY expires_at ASC\n LIMIT 1\n )\n RETURNING token", - "describe": { - "columns": [ - { - "name": "token", - "ordinal": 0, - "type_info": "Blob", - "origin": { - "Table": { - "table": "privacy_pass_tokens", - "name": "token" - } - } - } - ], - "parameters": { - "Right": 1 - }, - "nullable": [ - false - ] - }, - "hash": "f4af415f434d17c2880be292539d72ae5c0b8d1176a6c9860716832c1d82885b" -} diff --git a/rust/.sqlx/query-fbc5fa19cad3ddde6098197c6d58d103efaa97f89c7a3fa9d5ed4c8292a246a4.json b/rust/.sqlx/query-fbc5fa19cad3ddde6098197c6d58d103efaa97f89c7a3fa9d5ed4c8292a246a4.json new file mode 100644 index 00000000..4d623ce4 --- /dev/null +++ b/rust/.sqlx/query-fbc5fa19cad3ddde6098197c6d58d103efaa97f89c7a3fa9d5ed4c8292a246a4.json @@ -0,0 +1,38 @@ +{ + "db_name": "SQLite", + "query": "SELECT avatar_svg_compressed, sender_profile_counter FROM contacts WHERE user_id = ?", + "describe": { + "columns": [ + { + "name": "avatar_svg_compressed", + "ordinal": 0, + "type_info": "Blob", + "origin": { + "Table": { + "table": "contacts", + "name": "avatar_svg_compressed" + } + } + }, + { + "name": "sender_profile_counter", + "ordinal": 1, + "type_info": "Integer", + "origin": { + "Table": { + "table": "contacts", + "name": "sender_profile_counter" + } + } + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + true, + false + ] + }, + "hash": "fbc5fa19cad3ddde6098197c6d58d103efaa97f89c7a3fa9d5ed4c8292a246a4" +} diff --git a/rust/build.rs b/rust/build.rs index 97c47d40..3b1201be 100644 --- a/rust/build.rs +++ b/rust/build.rs @@ -1,17 +1,17 @@ use std::io::Result; fn main() -> Result<()> { + // The version the server compares against its minimum version setting. + // A build that cannot determine it has to fail here rather than ship a + // placeholder, which the server would read as an ancient client. println!("cargo:rerun-if-changed=../pubspec.yaml"); - if let Ok(pubspec) = std::fs::read_to_string("../pubspec.yaml") { - if let Some(version) = pubspec - .lines() - .find_map(|line| line.strip_prefix("version: ")) - { - println!( - "cargo:rustc-env=TWONLY_APP_VERSION={}", - version.split('+').next().unwrap_or(version) - ); - } - } + let pubspec = + std::fs::read_to_string("../pubspec.yaml").expect("could not read ../pubspec.yaml"); + let version = pubspec + .lines() + .find_map(|line| line.strip_prefix("version: ")) + .map(|version| version.split('+').next().unwrap_or(version).trim()) + .expect("no `version:` entry in ../pubspec.yaml"); + println!("cargo:rustc-env=TWONLY_APP_VERSION={version}"); let websocket_proto_root = "src/api/proto"; let websocket_protos = [ "src/api/proto/api/websocket/client_to_server.proto", diff --git a/rust/src/api/messages/incoming.rs b/rust/src/api/messages/incoming.rs index 23a57dc0..c971aec8 100644 --- a/rust/src/api/messages/incoming.rs +++ b/rust/src/api/messages/incoming.rs @@ -90,10 +90,12 @@ pub(crate) async fn handle_server_message( /// left out is redelivered on the next drain. /// /// Deduplication is the `received_receipts` claim inside -/// [`handle_decoded_server_message`]: it is committed in the same transaction -/// that persists the message, is keyed on the end-to-end receipt ID, and is -/// never purged. A redelivered envelope is therefore recognised before it is -/// decrypted, whichever transport carried it. +/// [`handle_decoded_server_message`]: it is keyed on the end-to-end receipt ID +/// and is never purged, so a redelivered envelope is recognised before it is +/// decrypted, whichever transport carried it. An encrypted message commits its +/// claim together with its plaintext one step ahead of the message itself, +/// because the ratchet step decryption spends cannot be rolled back; a retry +/// that finds that plaintext resumes from it instead of decrypting again. async fn acknowledge_pending_messages( ctx: &Arc, batch: server_to_client::PendingMessagesV2, @@ -114,14 +116,18 @@ async fn acknowledge_pending_messages( match result { // Committed, or recognised as a duplicate. Either way it is durable. Ok(()) => delivery_ids.push(delivery_id), - // An envelope that cannot be decoded will never decode. Acknowledge - // it so one poisoned row cannot be redelivered forever. + // An envelope that cannot be decoded, or content this client will + // never be able to act on, stays broken however often it is + // redelivered. Acknowledge it so one poisoned row cannot be + // redelivered forever. Err( - error @ (TwonlyError::ProtobufDecode(_) | TwonlyError::UnknownProtobufEnumValue(_)), + error @ (TwonlyError::ProtobufDecode(_) + | TwonlyError::UnknownProtobufEnumValue(_) + | TwonlyError::UnprocessableContent(_)), ) => { tracing::warn!( delivery_id, - "dropping an undecodable mailbox message: {error}" + "dropping an unprocessable mailbox message: {error}" ); delivery_ids.push(delivery_id); } @@ -225,7 +231,6 @@ async fn upgrade_legacy_session_to_v2( ) .execute(&database.pool) .await?; - database.notify_committed(["contacts"]); Ok(()) } @@ -247,7 +252,6 @@ pub(crate) async fn handle_decoded_server_message( if let Ok(user) = ctx.user_id().await { tracing::Span::current().record("user", user); } - tracing::info!("Started processing incoming message"); if message.receipt_id.is_empty() { return Err(TwonlyError::Generic( @@ -260,10 +264,14 @@ pub(crate) async fn handle_decoded_server_message( message_type, Type::Ciphertext | Type::PrekeyBundle | Type::CiphertextV2 ); - tracing::info!(is_encrypted_message, ?message_type, "Parsed message type"); + + tracing::info!( + is_encrypted_message, + ?message_type, + "Parsed incoming message type" + ); if is_encrypted_message { - tracing::info!("Ensuring contact exists..."); ensure_contact_exists(ctx, from_user_id).await?; } @@ -280,7 +288,20 @@ pub(crate) async fn handle_decoded_server_message( let claimed = Receipt::claim_received(&mut t, &message.receipt_id).await?; - if !claimed { + // A claim that was committed with a plaintext still parked on it belongs to + // a message whose handling did not commit. Its ratchet step is spent, so the + // redelivery has to resume from that plaintext instead of decrypting again. + let resumed_plaintext = if claimed || message_type != Type::CiphertextV2 { + None + } else { + Receipt::pending_plaintext(&mut t, &message.receipt_id).await? + }; + + if resumed_plaintext.is_some() { + tracing::info!("Resuming a redelivered message from its parked plaintext"); + } + + if !claimed && resumed_plaintext.is_none() { // Delivery receipts are terminal messages and must never themselves be // acknowledged. For regular messages Dart retries the delivery receipt // after ten days, atomically claiming the retry by moving created_at. @@ -315,19 +336,34 @@ pub(crate) async fn handle_decoded_server_message( sends_error_response = true; } Type::CiphertextV2 => { - let ciphertext = message.encrypted_content.ok_or_else(|| { - TwonlyError::Generic("V2 encrypted client message has no ciphertext".into()) - })?; - let decrypted = { - let engine = ctx.signal_engine.lock().await; - engine - .as_ref() - .ok_or(TwonlyError::SignalIdentityNotFound)? - .decrypt_message(from_user_id.to_string(), 1, ciphertext) - .await + let decrypted = match resumed_plaintext { + Some(plaintext) => Ok(plaintext), + None => { + let ciphertext = message.encrypted_content.ok_or_else(|| { + TwonlyError::UnprocessableContent( + "V2 encrypted client message has no ciphertext".into(), + ) + })?; + let engine = ctx.signal_engine.lock().await; + engine + .as_ref() + .ok_or(TwonlyError::SignalIdentityNotFound)? + .decrypt_message(from_user_id.to_string(), 1, ciphertext) + .await + } }; match decrypted { Ok(plaintext) => { + // Decryption consumed a ratchet step in the signal database, + // which this transaction cannot roll back. Commit the + // plaintext with the claim first, so a rollback further down + // leaves a redelivery something to resume from instead of a + // session that can no longer decrypt the message. + Receipt::store_pending_plaintext(&mut t, &message.receipt_id, &plaintext) + .await?; + t.commit().await?; + t = database.pool.begin().await?; + // Decrypting proves the peer speaks v2, so drop any 'v1' // marking left by a contact lookup that ran before the peer // published a prekey bundle. Otherwise every receipt back to @@ -350,7 +386,7 @@ pub(crate) async fn handle_decoded_server_message( let content = proto::EncryptedContent::decode(plaintext.as_slice()).map_err(|error| { - TwonlyError::Generic(format!( + TwonlyError::UnprocessableContent(format!( "invalid decrypted client content: {error}" )) })?; @@ -389,19 +425,13 @@ pub(crate) async fn handle_decoded_server_message( queue_sender_delivery_receipt(&mut t, from_user_id, &message.receipt_id).await?; } + // The message is about to become durable, so the plaintext parked for a + // retry that is no longer needed can go. + Receipt::clear_pending_plaintext(&mut t, &message.receipt_id).await?; + t.commit().await?; ctx.mark_incoming_committed(); - database.notify_committed([ - "received_receipts", - "receipts", - "messages", - "groups", - "contacts", - "key_verifications", - "user_discovery_own_promotions", - "notification_outbox", - ]); let ctx = ctx.clone(); tokio::spawn(async move { @@ -512,9 +542,15 @@ async fn handle_encrypted_inner( return recovery::handle_passwordless_recovery_heartbeat(t, from_user_id, heartbeat).await; } - let group_id = content - .group_id - .ok_or_else(|| TwonlyError::Generic("group-scoped message has no group ID".into()))?; + let Some(group_id) = content.group_id else { + // Everything below is group-scoped. Reaching here without a group ID is + // normal for a content that only carries sender metadata (a profile + // counter, a user-discovery version, a friend-promotion request), all of + // which has already been applied above. Failing it would only make the + // server redeliver a message there is nothing left to do with. + tracing::info!("Incoming message carried sender metadata only"); + return Ok(()); + }; if let Some(create) = content.group_create { return groups::handle_group_create(ctx, t, from_user_id, &group_id, create).await; @@ -610,7 +646,7 @@ async fn handle_encrypted_inner( .await; } - Err(TwonlyError::Generic(format!( + Err(TwonlyError::UnprocessableContent(format!( "client2client content in receipt {receipt_id} is not implemented in Rust" ))) } diff --git a/rust/src/api/messages/incoming/messages.rs b/rust/src/api/messages/incoming/messages.rs index 578d8627..04dc93dd 100644 --- a/rust/src/api/messages/incoming/messages.rs +++ b/rust/src/api/messages/incoming/messages.rs @@ -216,7 +216,6 @@ pub(crate) async fn ensure_contact_exists(ctx: &Arc, from_user_id: i64) .await?; } - db_app.notify_committed(["contacts"]); Ok(()) } @@ -448,7 +447,6 @@ pub(crate) async fn send_queued_receipt(ctx: &Arc, receipt_id: &str) -> // on every retry. Drop the receipt instead of queueing it forever. if row.account_deleted != 0 { Receipt::delete(&app_db.pool, receipt_id).await?; - app_db.notify_committed(["receipts"]); return Ok(()); } @@ -528,7 +526,6 @@ pub(crate) async fn send_queued_receipt(ctx: &Arc, receipt_id: &str) -> .await?; } t.commit().await?; - app_db.notify_committed(["receipts", "message_actions", "messages"]); Ok(()) } @@ -552,7 +549,6 @@ async fn defer_receipt_until_session( ) .execute(&database.pool) .await?; - database.notify_committed(["receipts"]); Ok(()) } diff --git a/rust/src/api/messages/outgoing.rs b/rust/src/api/messages/outgoing.rs index 9067a3ff..7ed135d3 100644 --- a/rust/src/api/messages/outgoing.rs +++ b/rust/src/api/messages/outgoing.rs @@ -129,7 +129,6 @@ pub async fn send_c2c_message_to_contact( .await?; t.commit().await?; - db_app.notify_committed(["receipts"]); if only_return_encrypted_data { return messages::prepare_queued_receipt(ctx, &receipt_id).await; diff --git a/rust/src/api/runtime.rs b/rust/src/api/runtime.rs index 38f02c3f..bfd6c2e3 100644 --- a/rust/src/api/runtime.rs +++ b/rust/src/api/runtime.rs @@ -145,7 +145,6 @@ impl ApiRuntime { } } } - database.notify_committed(["api_outbox"]); Ok(()) } diff --git a/rust/src/api/runtime/auth.rs b/rust/src/api/runtime/auth.rs index f5452b95..975c4b54 100644 --- a/rust/src/api/runtime/auth.rs +++ b/rust/src/api/runtime/auth.rs @@ -5,6 +5,7 @@ use super::client::ApiClient; use super::helpers::{decode_ok, schedule_post_authentication}; +use super::request::permanent_rejection_kind; use crate::api::proto::{client_to_server, server_to_client}; use crate::bridge::api::{ApiConnectionState, ApiEvent, ApiEventKind, ServerResult}; use crate::context::Context; @@ -21,6 +22,14 @@ use stream_tungstenite::handshake::{HandshakeReceiver, HandshakeSender, Handshak use stream_tungstenite::tokio_tungstenite::tungstenite; use tokio::sync::broadcast; +/// The app version from `pubspec.yaml` without the build number, baked in by +/// `build.rs`. This is what the handshake reports to the server, which compares +/// it against its configured minimum version. +/// +/// Not to be confused with `UserConfig::app_version`, which is the local +/// migration id and stays a purely client side concern. +pub(crate) const APP_VERSION: &str = env!("TWONLY_APP_VERSION"); + pub(crate) struct ApiAuthHandshaker { pub context: Arc, pub api_client: Weak, @@ -34,8 +43,7 @@ impl ApiAuthHandshaker { let user = UserConfig::load_from(&self.context)? .ok_or_else(|| TwonlyError::Generic("User configuration not found".into()))?; let device_id = user.device_id; - let app_version = user.app_version.to_string(); - Ok((Some(user.user_id), device_id, app_version)) + Ok((Some(user.user_id), device_id, APP_VERSION.to_string())) } async fn request_handshake( @@ -76,6 +84,17 @@ impl ApiAuthHandshaker { let ok = match ok_res { ServerResult::Ok(val) => val, ServerResult::ErrorCode(code) => { + // The handshake never reaches `handle_api_error`, so a rejection + // that no reconnect can fix has to be published here — otherwise + // the app just keeps failing to connect without telling the user + // that it is outdated or was logged out. + if let Some(kind) = permanent_rejection_kind(code) { + if let Some(client) = self.api_client.upgrade() { + // Closing the socket has to wait for this handshake to + // return, so it cannot be awaited from inside it. + tokio::spawn(async move { client.reject_permanently(kind).await }); + } + } return Err(HandshakeError::Protocol(format!( "Server returned error code: {}", code @@ -286,6 +305,7 @@ impl ApiAuthHandshaker { self.is_authenticated.store(true, Ordering::Release); if let Some(client) = self.api_client.upgrade() { + client.note_authenticated(); tokio::spawn(async move { client.set_state(ApiConnectionState::Authenticated).await; // Runs on every (re)connect, so this covers both the initial @@ -420,3 +440,42 @@ impl ApiClient { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::APP_VERSION; + + /// The server splits the reported version on '.' and parses every segment + /// as a number, so a suffix like "0.5.2-beta" would silently be read as + /// 0.5.0 and could fall below the configured minimum version. + #[test] + fn app_version_is_a_plain_dotted_version() { + let segments: Vec<&str> = APP_VERSION.split('.').collect(); + + assert!( + !segments.is_empty() && segments.len() <= 3, + "unexpected app version {APP_VERSION}" + ); + for segment in segments { + assert!( + segment.parse::().is_ok(), + "app version {APP_VERSION} has a non numeric segment {segment}" + ); + } + } + + /// Guards the `pubspec.yaml` parsing in `build.rs`. + #[test] + fn app_version_matches_the_pubspec() { + let pubspec = std::fs::read_to_string("../pubspec.yaml").expect("read ../pubspec.yaml"); + let version = pubspec + .lines() + .find_map(|line| line.strip_prefix("version: ")) + .expect("no `version:` entry in ../pubspec.yaml"); + + assert_eq!( + APP_VERSION, + version.split('+').next().unwrap_or(version).trim() + ); + } +} diff --git a/rust/src/api/runtime/client.rs b/rust/src/api/runtime/client.rs index cba471ca..aa2f618a 100644 --- a/rust/src/api/runtime/client.rs +++ b/rust/src/api/runtime/client.rs @@ -7,7 +7,7 @@ use crate::bridge::api::{ApiConfig, ApiConnectionState, ApiEvent, ApiEventKind}; use crate::context::Context; use crate::error::Result; use std::collections::HashMap; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, LazyLock, Weak}; use std::time::Duration; use stream_tungstenite::{ClientConfig, WebSocketClient}; @@ -29,6 +29,14 @@ const CATCH_UP_JITTER: Duration = Duration::from_secs(15); /// messages. const RECONNECT_INITIAL_DELAY: Duration = Duration::from_millis(250); const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(20); +/// A socket whose transport has died does not reconnect by itself: a failed +/// handshake makes the supervisor give up for good, and it leaves the send +/// channel behind so every later send reports a closed channel. These bound +/// the redial we drive ourselves, backing off so a handshake that keeps +/// failing (a rejected token, say) does not turn into a dial loop. +const FORCED_RECONNECT_INITIAL_DELAY: Duration = Duration::from_secs(1); +const FORCED_RECONNECT_MAX_DELAY: Duration = Duration::from_secs(60); + const RECEIVE_TIMEOUT: Duration = Duration::from_secs(45); const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); const HANDSHAKE_RETRY_DELAY: Duration = Duration::from_secs(1); @@ -53,6 +61,12 @@ pub(crate) struct ApiClient { /// `try_send`, so repeated triggers collapse into a single pull instead of /// queueing one request each. catch_up_tx: Mutex>>, + /// Set while a forced redial is pending, so the many sends that fail + /// against one dead socket schedule a single reconnect between them. + forced_reconnect_in_flight: AtomicBool, + /// Counts forced redials since the last authenticated session; feeds the + /// backoff in [`ApiClient::schedule_reconnect`]. + forced_reconnect_attempts: AtomicU32, } impl ApiClient { @@ -71,6 +85,8 @@ impl ApiClient { network_available: AtomicBool::new(true), is_authenticated: Arc::new(AtomicBool::new(false)), catch_up_tx: Mutex::const_new(None), + forced_reconnect_in_flight: AtomicBool::new(false), + forced_reconnect_attempts: AtomicU32::new(0), }) } @@ -143,6 +159,101 @@ impl ApiClient { }); } + /// Discards a socket the transport has declared dead and dials a fresh one. + /// + /// The supervisor cannot do this itself: it treats every failure our + /// handshaker reports as permanent, so it stops reconnecting, and because + /// that path skips its session cleanup it leaves the send channel in place + /// with no reader behind it. `ws_client` therefore still holds a client + /// that looks alive while every send fails with `ChannelClosed`, and + /// `connect` returns early on it. Only replacing it recovers. + /// + /// `stale` is the connection the caller failed on; if the slot already + /// holds a different one, someone reconnected in the meantime and this is + /// a no-op. The redial runs detached so the caller can return its own + /// error right away — whatever failed to send is retried by the + /// post-authentication receipt sweep once the new socket is up. + pub(crate) fn schedule_reconnect(self: &Arc, stale: &Arc, reason: &str) { + if self.deliberately_closed.load(Ordering::Acquire) + || API_PERMANENTLY_REJECTED.load(Ordering::Acquire) + { + return; + } + // Every send against the dead socket lands here; only the first one + // gets to schedule the redial. + if self.forced_reconnect_in_flight.swap(true, Ordering::AcqRel) { + return; + } + + let client = self.clone(); + let stale = stale.clone(); + let reason = reason.to_owned(); + tokio::spawn(async move { + let result = client.reconnect(&stale, &reason).await; + client + .forced_reconnect_in_flight + .store(false, Ordering::Release); + if let Err(error) = result { + tracing::warn!("reconnect after a dead API WebSocket failed: {error}"); + } + }); + } + + async fn reconnect(self: &Arc, stale: &Arc, reason: &str) -> Result<()> { + { + let mut guard = self.ws_client.lock().await; + if !guard + .as_ref() + .is_some_and(|current| Arc::ptr_eq(current, stale)) + { + return Ok(()); + } + guard.take(); + } + + // Counted only for a socket that was still the live one, so a stale + // report cannot inflate the backoff. + let attempt = self + .forced_reconnect_attempts + .fetch_add(1, Ordering::AcqRel); + let delay = FORCED_RECONNECT_INITIAL_DELAY + .saturating_mul(1_u32 << attempt.min(6)) + .min(FORCED_RECONNECT_MAX_DELAY); + tracing::warn!( + reason, + attempt, + ?delay, + "API WebSocket is dead, reconnecting" + ); + + self.is_authenticated.store(false, Ordering::Release); + // Dropping the sender ends the catch-up loop bound to the dead socket. + *self.catch_up_tx.lock().await = None; + if let Err(error) = stale.shutdown_graceful(Duration::from_secs(5)).await { + tracing::warn!("dead WebSocket did not shut down cleanly: {error}"); + } + self.fail_pending().await; + self.set_state(ApiConnectionState::Stopped).await; + + tokio::time::sleep(delay).await; + + // Closing the client or losing the network during the delay means the + // redial is no longer wanted. + if self.deliberately_closed.load(Ordering::Acquire) + || !self.network_available.load(Ordering::Acquire) + { + return Ok(()); + } + self.connect().await + } + + /// Restarts the forced-reconnect backoff. A session that got all the way to + /// an authenticated handshake proves the credentials and the server are + /// fine, so the next dead socket starts over at the short delay. + pub(crate) fn note_authenticated(&self) { + self.forced_reconnect_attempts.store(0, Ordering::Release); + } + pub(crate) async fn set_state(&self, state: ApiConnectionState) { let mut guard = self.state.write().await; if *guard != state { @@ -221,6 +332,7 @@ impl ApiClient { }); let self_clone = self.clone(); + let connection = Arc::downgrade(&ws_arc); tokio::spawn(async move { loop { tokio::select! { @@ -258,6 +370,20 @@ impl ApiClient { Ok(ConnectionEvent::Connecting { .. }) => { self_clone.set_state(ApiConnectionState::Connecting).await; } + // The supervisor has stopped reconnecting: it + // treats every handshake failure as permanent, and + // that path leaves the send channel behind with no + // reader, so this socket can neither reconnect nor + // send. Replace it instead of sitting on it. + Ok(ConnectionEvent::FatalError { .. } | ConnectionEvent::Shutdown) => { + self_clone.is_authenticated.store(false, Ordering::Release); + if let Some(connection) = connection.upgrade() { + self_clone.schedule_reconnect( + &connection, + "supervisor stopped reconnecting", + ); + } + } Ok(_) => {} Err(_) => break, } diff --git a/rust/src/api/runtime/request.rs b/rust/src/api/runtime/request.rs index 7baee67c..eaf6228e 100644 --- a/rust/src/api/runtime/request.rs +++ b/rust/src/api/runtime/request.rs @@ -30,6 +30,29 @@ fn call_handle_server_message( }) } +/// Maps the error codes that make further connection attempts pointless onto +/// the event the UI shows for them. Everything else is a per-request error. +pub(crate) fn permanent_rejection_kind(code: i32) -> Option { + use crate::api::proto::error::ErrorCode; + if code == ErrorCode::AppVersionOutdated as i32 { + Some(ApiEventKind::AppOutdated) + } else if code == ErrorCode::NewDeviceRegistered as i32 { + Some(ApiEventKind::NewDeviceRegistered) + } else { + None + } +} + +/// Whether a send failure means the socket can never carry another frame. +/// `ChannelClosed` is the signature of a connection the supervisor abandoned +/// mid-handshake: it left the send channel in place but dropped the reader, so +/// every further send fails the same way until the client is replaced. The +/// other variants describe a single message or a connection that is still +/// coming up, and are left to the caller to retry. +fn is_dead_transport(error: &SendError) -> bool { + matches!(error, SendError::ChannelClosed) +} + impl ApiClient { pub(crate) async fn handle_incoming(self: &Arc, bytes: &[u8]) { let Ok(message) = server_to_client::ServerToClient::decode(bytes) else { @@ -90,7 +113,12 @@ impl ApiClient { Err(SendError::NotConnected) if start.elapsed() < Duration::from_secs(10) => { tokio::time::sleep(Duration::from_millis(50)).await; } - Err(e) => return Err(TwonlyError::Generic(format!("send error: {:?}", e))), + Err(e) => { + if is_dead_transport(&e) { + self.schedule_reconnect(&client, "send failed"); + } + return Err(TwonlyError::Generic(format!("send error: {:?}", e))); + } } } } else { @@ -103,7 +131,7 @@ impl ApiClient { } pub(crate) async fn request_durable( - &self, + self: &Arc, bytes: Vec, operation_kind: &str, ) -> Result> { @@ -129,13 +157,12 @@ impl ApiClient { ) .execute(&database.pool) .await?; - database.notify_committed(["api_outbox"]); } response } pub(crate) async fn request_internal( - &self, + self: &Arc, bytes: Vec, timeout: Duration, ) -> Result> { @@ -167,6 +194,9 @@ impl ApiClient { } Err(e) => { self.pending.lock().await.remove(&sequence); + if is_dead_transport(&e) { + self.schedule_reconnect(&client, "request send failed"); + } return Err(TwonlyError::Generic(format!("send error: {:?}", e))); } } @@ -218,30 +248,30 @@ impl ApiClient { Ok(()) } + /// Rejections the server will keep returning for as long as this + /// installation stays as it is: the socket is closed for good and the UI is + /// told why, instead of the client reconnecting into the same error. + pub(crate) async fn reject_permanently(&self, kind: ApiEventKind) { + self.set_state(ApiConnectionState::PermanentlyRejected) + .await; + API_PERMANENTLY_REJECTED.store(true, Ordering::Release); + self.deliberately_closed.store(true, Ordering::Release); + let _ = self.events.send(ApiEvent { + kind, + state: Some(ApiConnectionState::PermanentlyRejected), + message: None, + }); + if let Some(client) = self.ws_client.lock().await.take() { + if let Err(error) = client.shutdown_graceful(Duration::from_secs(5)).await { + tracing::warn!(%error, "permanently rejected WebSocket did not shut down cleanly"); + } + } + } + pub(crate) async fn handle_api_error(&self, code: i32, contact_id: Option) -> Result<()> { use crate::api::proto::error::ErrorCode; - if code == ErrorCode::AppVersionOutdated as i32 - || code == ErrorCode::NewDeviceRegistered as i32 - { - self.set_state(ApiConnectionState::PermanentlyRejected) - .await; - API_PERMANENTLY_REJECTED.store(true, Ordering::Release); - self.deliberately_closed.store(true, Ordering::Release); - let kind = if code == ErrorCode::AppVersionOutdated as i32 { - ApiEventKind::AppOutdated - } else { - ApiEventKind::NewDeviceRegistered - }; - let _ = self.events.send(ApiEvent { - kind, - state: Some(ApiConnectionState::PermanentlyRejected), - message: None, - }); - if let Some(client) = self.ws_client.lock().await.take() { - if let Err(error) = client.shutdown_graceful(Duration::from_secs(5)).await { - tracing::warn!(%error, "permanently rejected WebSocket did not shut down cleanly"); - } - } + if let Some(kind) = permanent_rejection_kind(code) { + self.reject_permanently(kind).await; } if code == ErrorCode::UserIdNotFound as i32 { if let Some(contact_id) = contact_id { @@ -258,7 +288,6 @@ impl ApiClient { .execute(&mut *transaction) .await?; transaction.commit().await?; - database.notify_committed(["contacts", "receipts"]); } } Ok(()) diff --git a/rust/src/api/server/contacts.rs b/rust/src/api/server/contacts.rs index 2db814fe..0e670b55 100644 --- a/rust/src/api/server/contacts.rs +++ b/rust/src/api/server/contacts.rs @@ -61,7 +61,6 @@ impl Server { .await?; } } - database.notify_committed(["contacts"]); Ok(()) } diff --git a/rust/src/bridge/wrapper/user_discovery.rs b/rust/src/bridge/wrapper/user_discovery.rs index b0afc93c..efa686ec 100644 --- a/rust/src/bridge/wrapper/user_discovery.rs +++ b/rust/src/bridge/wrapper/user_discovery.rs @@ -76,7 +76,6 @@ impl FlutterUserDiscovery { .execute(&mut *transaction) .await?; transaction.commit().await?; - database.notify_committed(["user_discovery_own_promotions", "contacts"]); Ok(()) }) .await diff --git a/rust/src/database/app/legacy_import.rs b/rust/src/database/app/legacy_import.rs index 5a4f6b74..4575397f 100644 --- a/rust/src/database/app/legacy_import.rs +++ b/rust/src/database/app/legacy_import.rs @@ -66,7 +66,6 @@ impl AppDatabase { .execute(&mut *connection) .await; let report = import_result?; - self.notify_committed(APPLICATION_TABLES.iter().copied()); Ok(report) } diff --git a/rust/src/database/app/migrations/0008_pending_plaintext.sql b/rust/src/database/app/migrations/0008_pending_plaintext.sql new file mode 100644 index 00000000..267c17be --- /dev/null +++ b/rust/src/database/app/migrations/0008_pending_plaintext.sql @@ -0,0 +1,12 @@ +-- Signal decryption advances the double ratchet in the *signal* database, which +-- has nothing to do with the app-database transaction that guards the receipt +-- claim. Rolling that transaction back after a successful decryption therefore +-- un-claims the receipt without rewinding the ratchet: the server redelivers +-- the envelope and the retry dies with "message with old counter", losing the +-- message for good. +-- +-- The plaintext is parked here in the same commit that claims the receipt, so a +-- retry resumes from it instead of decrypting a second time. It is cleared as +-- soon as the message has been handled. +ALTER TABLE received_receipts + ADD COLUMN pending_plaintext BLOB; diff --git a/rust/src/database/app/mod.rs b/rust/src/database/app/mod.rs index 8b72a328..6962b396 100644 --- a/rust/src/database/app/mod.rs +++ b/rust/src/database/app/mod.rs @@ -75,10 +75,15 @@ impl AppDatabase { let (changes, _) = broadcast::channel(256); // SQLite itself reports which tables a statement touched, so Drift's - // query streams stay correct without every Rust write site having to - // remember an explicit `notify_committed`. Rows are collected as they - // are written and only published once the transaction commits, so a - // rollback never reaches the UI. + // query streams stay correct without any write site having to announce + // what it changed. Rows are collected as they are written and only + // published once the transaction commits, so a rollback never reaches + // the UI. + // + // The one write this cannot see is `DELETE FROM ` with no + // `WHERE`: SQLite's truncate optimization drops the rows without + // invoking the update hook. Give such a delete a `WHERE 1` so it takes + // the ordinary path. let pending: Arc>> = Arc::default(); let hook_pending = pending.clone(); let hook_changes = changes.clone(); @@ -154,17 +159,6 @@ impl AppDatabase { self.changes.subscribe() } - /// Publishes a change immediately, without waiting for a commit. - /// - /// The SQLite hooks installed in [`AppDatabase::new`] already cover every - /// ordinary write. This stays for the cases they cannot see -- most notably - /// `DELETE FROM
` with no `WHERE`, which SQLite's truncate - /// optimization performs without invoking the update hook. - pub fn notify_committed<'a>(&self, tables: impl IntoIterator) { - let tables = tables.into_iter().map(str::to_owned).collect(); - let _ = self.changes.send(DatabaseChange { tables }); - } - pub async fn raw_select(&self, statement: String, arguments: Vec) -> Result { let rows = bind_arguments(sqlx::query(AssertSqlSafe(statement)), arguments)? .fetch_all(&self.pool) diff --git a/rust/src/database/app/tables/receipt.rs b/rust/src/database/app/tables/receipt.rs index 6889f428..b40de960 100644 --- a/rust/src/database/app/tables/receipt.rs +++ b/rust/src/database/app/tables/receipt.rs @@ -94,6 +94,71 @@ impl Receipt { Ok(claimed) } + /// Parks the plaintext of an already-decrypted message on its receipt + /// claim. Signal decryption is a one-shot operation — it advances the + /// double ratchet in the signal database, outside this transaction — so the + /// plaintext has to be committed together with the claim. A redelivery that + /// finds it here resumes handling instead of decrypting again, which would + /// fail with an old-counter error. + pub async fn store_pending_plaintext( + transaction: &mut Transaction<'_, Sqlite>, + receipt_id: &str, + plaintext: &[u8], + ) -> Result<()> { + sqlx::query!( + r#" + UPDATE received_receipts + SET pending_plaintext = ? + WHERE receipt_id = ? + "#, + plaintext, + receipt_id, + ) + .execute(&mut **transaction) + .await?; + + Ok(()) + } + + /// Returns the parked plaintext of a receipt whose handling did not commit. + pub async fn pending_plaintext( + transaction: &mut Transaction<'_, Sqlite>, + receipt_id: &str, + ) -> Result>> { + let plaintext = sqlx::query_scalar!( + r#" + SELECT pending_plaintext + FROM received_receipts + WHERE receipt_id = ? + "#, + receipt_id, + ) + .fetch_optional(&mut **transaction) + .await? + .flatten(); + + Ok(plaintext) + } + + /// Drops a parked plaintext once its message no longer needs it. + pub async fn clear_pending_plaintext( + transaction: &mut Transaction<'_, Sqlite>, + receipt_id: &str, + ) -> Result<()> { + sqlx::query!( + r#" + UPDATE received_receipts + SET pending_plaintext = NULL + WHERE receipt_id = ? AND pending_plaintext IS NOT NULL + "#, + receipt_id, + ) + .execute(&mut **transaction) + .await?; + + Ok(()) + } + pub async fn claim_received_retry( transaction: &mut Transaction<'_, Sqlite>, receipt_id: &str, @@ -201,3 +266,58 @@ impl<'a> NewReceipt<'a> { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::database::app::AppDatabase; + + #[tokio::test] + async fn a_parked_plaintext_survives_a_rolled_back_handling() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("app.sqlite"); + let database = AppDatabase::new(path.to_str().unwrap(), None, false) + .await + .unwrap(); + database.run_migrations().await.unwrap(); + + // Phase one: the claim and the plaintext of a message whose ratchet step + // is already spent are committed together. + let mut t = database.pool.begin().await.unwrap(); + assert!(Receipt::claim_received(&mut t, "receipt").await.unwrap()); + Receipt::store_pending_plaintext(&mut t, "receipt", b"decrypted") + .await + .unwrap(); + t.commit().await.unwrap(); + + // Phase two fails, so nothing it wrote survives. + let mut t = database.pool.begin().await.unwrap(); + Receipt::clear_pending_plaintext(&mut t, "receipt") + .await + .unwrap(); + t.rollback().await.unwrap(); + + // The redelivery is not a fresh claim, but it can resume without + // decrypting the message a second time. + let mut t = database.pool.begin().await.unwrap(); + assert!(!Receipt::claim_received(&mut t, "receipt").await.unwrap()); + assert_eq!( + Receipt::pending_plaintext(&mut t, "receipt").await.unwrap(), + Some(b"decrypted".to_vec()) + ); + + // Once handling commits, the plaintext is gone and a later redelivery is + // recognised as the plain duplicate it is. + Receipt::clear_pending_plaintext(&mut t, "receipt") + .await + .unwrap(); + t.commit().await.unwrap(); + + let mut t = database.pool.begin().await.unwrap(); + assert!(!Receipt::claim_received(&mut t, "receipt").await.unwrap()); + assert_eq!( + Receipt::pending_plaintext(&mut t, "receipt").await.unwrap(), + None + ); + } +} diff --git a/rust/src/error.rs b/rust/src/error.rs index f84dc22c..426eb9cc 100644 --- a/rust/src/error.rs +++ b/rust/src/error.rs @@ -78,6 +78,13 @@ pub enum TwonlyError { #[error("encrypted media is {bytes} bytes but the plan allows {limit}")] MediaTooLarge { bytes: i64, limit: i64 }, + /// Content that will never be processable: it is malformed, or it uses a + /// feature this client does not implement. Callers must not retry it — + /// a reliable-mailbox row carrying such a message is acknowledged so the + /// server stops redelivering it forever. + #[error("unprocessable content: {0}")] + UnprocessableContent(String), + #[error("{0}")] Generic(String), diff --git a/rust/src/log.rs b/rust/src/log.rs index afcf71d4..8fd09dfa 100644 --- a/rust/src/log.rs +++ b/rust/src/log.rs @@ -110,6 +110,115 @@ impl<'a> MakeWriter<'a> for AppLogWriter { } } +/// Android sends a native library's stdout to `/dev/null`; only `liblog` +/// reaches logcat, so the console layer has to go through it there. +#[cfg(target_os = "android")] +mod logcat { + use std::ffi::CString; + use std::os::raw::{c_char, c_int}; + use tracing::Metadata; + use tracing_subscriber::fmt::MakeWriter; + + #[link(name = "log")] + extern "C" { + fn __android_log_write(prio: c_int, tag: *const c_char, text: *const c_char) -> c_int; + } + + const TAG: &[u8] = b"twonly\0"; + /// logcat truncates a record at ~4 KiB, so split before it does. + const MAX_PAYLOAD: usize = 3800; + + const PRIO_VERBOSE: c_int = 2; + const PRIO_DEBUG: c_int = 3; + const PRIO_INFO: c_int = 4; + const PRIO_WARN: c_int = 5; + const PRIO_ERROR: c_int = 6; + + #[derive(Clone, Copy)] + pub(super) struct LogcatWriter; + + pub(super) struct LogcatBuffer { + priority: c_int, + buffer: Vec, + } + + impl LogcatBuffer { + const fn new(priority: c_int) -> Self { + Self { + priority, + buffer: Vec::new(), + } + } + + fn emit(&self, message: &str) { + let Ok(text) = CString::new(message) else { + return; + }; + // SAFETY: both pointers are NUL-terminated and outlive the call. + unsafe { + __android_log_write(self.priority, TAG.as_ptr().cast::(), text.as_ptr()); + } + } + } + + impl std::io::Write for LogcatBuffer { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.buffer.extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl Drop for LogcatBuffer { + fn drop(&mut self) { + if self.buffer.is_empty() { + return; + } + let text = String::from_utf8_lossy(&self.buffer); + for line in text.lines() { + // An interior NUL would silently cut the record short. + let line = line.replace('\0', ""); + let mut rest = line.as_str(); + while !rest.is_empty() { + let split = if rest.len() <= MAX_PAYLOAD { + rest.len() + } else { + let mut index = MAX_PAYLOAD; + while index > 0 && !rest.is_char_boundary(index) { + index -= 1; + } + index.max(1) + }; + let (chunk, remainder) = rest.split_at(split); + self.emit(chunk); + rest = remainder; + } + } + } + } + + impl<'a> MakeWriter<'a> for LogcatWriter { + type Writer = LogcatBuffer; + + fn make_writer(&'a self) -> Self::Writer { + LogcatBuffer::new(PRIO_INFO) + } + + fn make_writer_for(&'a self, meta: &Metadata<'_>) -> Self::Writer { + LogcatBuffer::new(match *meta.level() { + tracing::Level::TRACE => PRIO_VERBOSE, + tracing::Level::DEBUG => PRIO_DEBUG, + tracing::Level::INFO => PRIO_INFO, + tracing::Level::WARN => PRIO_WARN, + tracing::Level::ERROR => PRIO_ERROR, + }) + } + } +} + #[derive(Clone, Copy, Debug, Default)] pub struct PlainFields; @@ -300,11 +409,19 @@ pub(crate) fn init_tracing(data_dir: &Path, in_background: bool) { } TRACING_INIT.get_or_init(|| { - let stdout_layer = Layer::new() + #[cfg(not(target_os = "android"))] + let console_layer = Layer::new() .with_writer(std::io::stdout) .with_ansi(false) .event_format(ShortEventFormatter::ansi()); + #[cfg(target_os = "android")] + let console_layer = Layer::new() + .with_writer(logcat::LogcatWriter) + .with_ansi(false) + .fmt_fields(PlainFields) + .event_format(ShortEventFormatter::plain()); + let default_filter = if std::env::var("FLUTTER_TEST").is_ok() { "info,refinery_core=warn,refinery=warn" } else { @@ -322,7 +439,7 @@ pub(crate) fn init_tracing(data_dir: &Path, in_background: bool) { EnvFilter::try_from_default_env() .unwrap_or_else(|_| EnvFilter::new(default_filter)), ) - .with(stdout_layer) + .with(console_layer) .with(file_layer) .try_init(); }); diff --git a/rust/src/native/video.rs b/rust/src/native/video.rs index 17727aee..856a1b95 100644 --- a/rust/src/native/video.rs +++ b/rust/src/native/video.rs @@ -61,7 +61,6 @@ fn report_progress(media_id: &str, percent: i64) { .bind(&media_id) .execute(&database.pool) .await; - database.notify_committed(["media_files"]); }); } diff --git a/rust/src/services/contacts.rs b/rust/src/services/contacts.rs index 2571cb53..6f64f122 100644 --- a/rust/src/services/contacts.rs +++ b/rust/src/services/contacts.rs @@ -52,7 +52,6 @@ impl ContactService { .insert_on_conflict_update(&mut transaction) .await?; transaction.commit().await?; - database.notify_committed(["contacts"]); self.send_contact_request( user.user_id, @@ -115,7 +114,6 @@ impl ContactService { .execute(&database.pool) .await?; release_deferred_receipts(&database, user_id).await?; - database.notify_committed(["contacts", "receipts"]); Ok(()) } @@ -223,7 +221,6 @@ impl ContactService { Group::create_direct_chat(&self.ctx, &mut transaction, contact).await?; transaction.commit().await?; - database.notify_committed(["contacts", "groups"]); self.send_contact_request( contact_id, @@ -258,7 +255,6 @@ impl ContactService { .await?; t.commit().await?; - db_app.notify_committed(["contacts"]); self.send_contact_request( contact_id, diff --git a/rust/src/services/direct_media_upload.rs b/rust/src/services/direct_media_upload.rs index 68cfe4cb..b060ac17 100644 --- a/rust/src/services/direct_media_upload.rs +++ b/rust/src/services/direct_media_upload.rs @@ -416,7 +416,6 @@ impl DirectMediaUploadService { .await?; } } - database.notify_committed(["media_files", "messages", "receipts"]); Ok(()) } diff --git a/rust/src/services/groups/mod.rs b/rust/src/services/groups/mod.rs index 2a73ce99..e064cb91 100644 --- a/rust/src/services/groups/mod.rs +++ b/rust/src/services/groups/mod.rs @@ -151,7 +151,6 @@ impl GroupService { Group::set_last_flame_sync(&db.pool, &group.group_id, now).await?; } - db.notify_committed(["groups"]); Ok(()) } @@ -224,7 +223,6 @@ impl GroupService { .await?; transaction.commit().await?; - database.notify_committed(["receipts", "groups", "group_members"]); Ok(()) } @@ -278,7 +276,6 @@ impl GroupService { } Self::history(&mut tr, &group_id, "createdGroup", None, None, None).await?; tr.commit().await?; - db.notify_committed(["groups", "group_members", "group_histories"]); MessageService::new(&self.ctx) .send_to_group( group_id.clone(), @@ -317,7 +314,6 @@ impl GroupService { .apply_fetched_group_state(&mut t, &group_id, server) .await?; t.commit().await?; - database.notify_committed(["groups", "group_members", "contacts", "receipts"]); // `apply_state` may have queued public-key requests. Nothing else // flushes them here -- this is not an inbound-message path -- and they @@ -659,7 +655,6 @@ impl GroupService { .execute(&mut tr) .await?; tr.commit().await?; - db.notify_committed(["groups", "group_histories"]); Ok(true) } @@ -687,7 +682,6 @@ impl GroupService { .insert_on_conflict_update(&mut tr) .await?; tr.commit().await?; - database.notify_committed(["contacts"]); Ok(true) } @@ -722,7 +716,6 @@ impl GroupService { self.broadcast_group_public_key(&mut transaction, group_id) .await?; transaction.commit().await?; - database.notify_committed(["receipts"]); Ok(()) } diff --git a/rust/src/services/media_upload.rs b/rust/src/services/media_upload.rs index c1eb43df..d41be700 100644 --- a/rust/src/services/media_upload.rs +++ b/rust/src/services/media_upload.rs @@ -138,7 +138,6 @@ impl MediaUploadService { .execute(&mut *transaction) .await?; transaction.commit().await?; - database.notify_committed(["media_files"]); Ok(media_id) } @@ -199,7 +198,6 @@ impl MediaUploadService { Group::record_media_exchange(&mut transaction, group_id, false, now).await?; } transaction.commit().await?; - database.notify_committed(["messages", "media_files", "groups", "contacts"]); drop(database); // Preparation compresses and encrypts, which is far too slow to keep the @@ -375,7 +373,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files", "receipts"]); drop(database); // The legacy pre-built upload request would pin the old recipient set. @@ -501,7 +498,6 @@ impl MediaUploadService { .bind(&receipt.receipt_id) .execute(&database.pool) .await?; - database.notify_committed(["messages", "receipts"]); return Ok(()); }; drop(database); @@ -573,7 +569,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files"]); Ok(()) } @@ -669,7 +664,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files", "messages"]); Ok(()) } @@ -689,7 +683,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files", "messages"]); MediaFileService::new(&self.ctx).remove_files(media_id, &media.media_type)?; tracing::warn!( media_id, @@ -865,7 +858,6 @@ impl MediaUploadService { .execute(&database.pool) .await?; } - database.notify_committed(["media_files"]); Ok(()) } @@ -877,7 +869,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files"]); Ok(()) } @@ -891,7 +882,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files"]); Ok(()) } @@ -920,7 +910,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files"]); return Ok(()); } @@ -935,7 +924,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files"]); return Ok(()); } @@ -950,7 +938,6 @@ impl MediaUploadService { .bind(media_id) .execute(&database.pool) .await?; - database.notify_committed(["media_files"]); Ok(()) } @@ -961,7 +948,6 @@ impl MediaUploadService { let mut transaction = database.pool.begin().await?; MediaFile::mark_uploaded(&mut transaction, media_id).await?; transaction.commit().await?; - database.notify_committed(["media_files", "messages", "receipts"]); Ok(()) } diff --git a/rust/src/services/mediafiles.rs b/rust/src/services/mediafiles.rs index 4deef12b..48b9e8e0 100644 --- a/rust/src/services/mediafiles.rs +++ b/rust/src/services/mediafiles.rs @@ -88,7 +88,6 @@ impl MediaFileService { ) .execute(&database.pool) .await?; - database.notify_committed(["media_files"]); } result } @@ -139,7 +138,6 @@ impl MediaFileService { if let Some(media_type) = media_type { self.remove_files(media_id, &media_type)?; } - database.notify_committed(["media_files"]); return Ok(()); } if messages.len() != 1 { @@ -171,7 +169,6 @@ impl MediaFileService { if let Some(media_type) = media_type { self.remove_files(media_id, &media_type)?; } - database.notify_committed(["messages", "media_files"]); return Ok(()); } @@ -226,7 +223,6 @@ impl MediaFileService { self.request_reupload(media_id).await?; return Ok(()); } - database.notify_committed(["media_files"]); Ok(()) } @@ -301,7 +297,6 @@ impl MediaFileService { .fetch_all(&database.pool) .await?; - database.notify_committed(["media_files"]); drop(database); for target in targets { diff --git a/rust/src/services/messages.rs b/rust/src/services/messages.rs index 9c960b32..3ab4778a 100644 --- a/rust/src/services/messages.rs +++ b/rust/src/services/messages.rs @@ -41,7 +41,6 @@ impl MessageService { { sqlx::query!("UPDATE groups SET last_message_exchange = CAST(strftime('%s','now') AS INTEGER) WHERE group_id = ?", group_id) .execute(&database.pool).await?; - database.notify_committed(["groups"]); } let members = sqlx::query_scalar!( r#"SELECT contact_id FROM group_members @@ -168,6 +167,16 @@ impl MessageService { Ok(()) } + /// Persists an outgoing text message and hands the delivery off. + /// + /// The composer is blocked on this call, and the chat list can only render + /// the new bubble once its own `SELECT` gets the single SQLite connection + /// back. Everything the send needs -- decorating the payload, queueing a + /// receipt per member, establishing a session, the server round trip -- + /// competes for exactly that connection, which is how a message could reach + /// the other side before showing up here. So the row the UI renders from is + /// committed and published on its own, in one transaction, and the send + /// runs on a separate task. pub async fn insert_and_send_text( &self, group_id: String, @@ -177,13 +186,14 @@ impl MessageService { let database = self.ctx.app_db.read().await.clone(); let message_id = uuid::Uuid::new_v4().to_string(); let timestamp = chrono::Utc::now().timestamp_millis(); + + let mut t = database.pool.begin().await?; sqlx::query!( "UPDATE groups SET draft_message = NULL WHERE group_id = ?", group_id ) - .execute(&database.pool) + .execute(&mut *t) .await?; - database.notify_committed(["groups"]); sqlx::query!( r#"INSERT INTO messages(group_id, message_id, type, content, quotes_message_id, created_at) VALUES (?, ?, 'text', ?, ?, ?)"#, @@ -193,25 +203,41 @@ impl MessageService { quote_message_id, timestamp / 1000, ) - .execute(&database.pool) + .execute(&mut *t) .await?; - database.notify_committed(["messages"]); - self.send_to_group( - group_id, - proto::EncryptedContent { - text_message: Some(encrypted_content::TextMessage { - sender_message_id: message_id.clone(), - text, - timestamp, - quote_message_id, - }), - ..Default::default() + // The commit hook installed on the connection publishes `groups` and + // `messages` on its own, so the chat list is already on its way. + t.commit().await?; + + let content = proto::EncryptedContent { + text_message: Some(encrypted_content::TextMessage { + sender_message_id: message_id.clone(), + text, + timestamp, + quote_message_id, + }), + ..Default::default() + } + .encode_to_vec(); + + // The message row already carries the "not acknowledged yet" state the + // bubble shows, so a failure here is reported the same way a failed + // network send is: the bubble stays unacknowledged until a retry sweep + // gets it through. + let ctx = self.ctx.clone(); + let sent_message_id = message_id.clone(); + tokio::spawn(async move { + if let Err(error) = Self::new(&ctx) + .send_to_group(group_id, content, Some(sent_message_id.clone()), false) + .await + { + tracing::warn!( + message_id = sent_message_id, + "sending the text message failed: {error}" + ); } - .encode_to_vec(), - Some(message_id.clone()), - false, - ) - .await?; + }); + Ok(message_id) } @@ -235,7 +261,6 @@ impl MessageService { ) .execute(&database.pool) .await?; - database.notify_committed(["messages"]); self.send_to_group( group_id, proto::EncryptedContent { @@ -330,7 +355,6 @@ impl MessageService { .ok_or_else(|| TwonlyError::Generic("contact does not exist".into()))?; Group::create_direct_chat(&self.ctx, &mut transaction, contact).await?; transaction.commit().await?; - database.notify_committed(["groups", "group_members"]); let data = proto::AdditionalMessageData { r#type: proto::additional_message_data::Type::AskAboutUser as i32, link: None, @@ -472,7 +496,6 @@ impl MessageService { ) .await?; transaction.commit().await?; - database.notify_committed(["messages", "notification_outbox"]); Ok(()) } diff --git a/rust/src/services/notifications.rs b/rust/src/services/notifications.rs index 86a072a5..b1fbafad 100644 --- a/rust/src/services/notifications.rs +++ b/rust/src/services/notifications.rs @@ -279,7 +279,7 @@ pub(crate) async fn record_incoming_event( /// stale rows keeps them from producing an alert or inflating the badge. async fn clear_stale_opened(database: &Arc) -> Result<()> { let cleared_at = current_time().timestamp(); - let cleared = sqlx::query( + sqlx::query( r#" UPDATE notification_outbox SET cleared_at = ? @@ -296,9 +296,6 @@ async fn clear_stale_opened(database: &Arc) -> Result<()> { .bind(cleared_at) .execute(&database.pool) .await?; - if cleared.rows_affected() != 0 { - database.notify_committed(["notification_outbox"]); - } Ok(()) } @@ -481,7 +478,6 @@ pub async fn acknowledge_batch(ctx: &Arc, event_ids: &[String]) -> Resu .await?; } transaction.commit().await?; - database.notify_committed(["notification_outbox"]); Ok(()) } @@ -503,7 +499,6 @@ pub async fn clear_conversation(ctx: &Arc, conversation_id: &str) -> Re .execute(&mut *transaction) .await?; transaction.commit().await?; - database.notify_committed(["notification_outbox"]); Ok(notification_ids) } @@ -531,9 +526,6 @@ pub async fn clear_contact_requests(ctx: &Arc) -> Result> { .execute(&mut *transaction) .await?; transaction.commit().await?; - if !notification_ids.is_empty() { - database.notify_committed(["notification_outbox"]); - } Ok(notification_ids) } diff --git a/rust/src/user_discovery.rs b/rust/src/user_discovery.rs index df5de504..b53e946b 100644 --- a/rust/src/user_discovery.rs +++ b/rust/src/user_discovery.rs @@ -119,7 +119,6 @@ impl UserDiscovery { .execute(&database.pool) .await?; } - database.notify_committed(["user_discovery_announced_users"]); Ok(()) } pub fn new( @@ -195,7 +194,6 @@ impl UserDiscovery { ) .await?; transaction.commit().await?; - database.notify_committed(["user_discovery_shares"]); Ok(()) } @@ -620,7 +618,11 @@ impl UserDiscovery { let split_index = shares.len() - (config.threshold - 1) as usize; verification_shares.extend(shares.drain(split_index..)); - sqlx::query!("DELETE FROM user_discovery_shares") + // `WHERE 1` is load-bearing: without a WHERE clause SQLite takes the + // truncate optimization, which drops every row without ever invoking + // the update hook the change stream is built on, so the UI would never + // learn the shares were replaced. + sqlx::query!("DELETE FROM user_discovery_shares WHERE 1") .execute(&mut **t) .await .map_err(|error| TwonlyError::UserDiscoveryStore(error.to_string()))?; diff --git a/rust/tests/api/tester.rs b/rust/tests/api/tester.rs index 15f48a57..37e02511 100644 --- a/rust/tests/api/tester.rs +++ b/rust/tests/api/tester.rs @@ -71,7 +71,6 @@ impl Tester { .execute(&database.pool) .await?; } - database.notify_committed(["key_verifications"]); Ok(()) } @@ -629,7 +628,6 @@ impl Tester { ) .execute(&database.pool) .await?; - database.notify_committed(["contacts"]); Ok(()) }