From f3f707899ccf5137b8c20545ee96cffe7d96baee Mon Sep 17 00:00:00 2001 From: Tryn McCaffery Date: Tue, 28 Jul 2026 17:43:31 -0400 Subject: [PATCH 1/5] fix(mobile): deliver DMs after reconnect Signed-off-by: Tryn McCaffery --- .../channels/send_message_provider.dart | 24 +++++++- .../profile/presence_cache_provider.dart | 58 ++++++++++++++++--- mobile/lib/shared/relay/relay_session.dart | 29 +++++++++- .../channels/send_message_provider_test.dart | 53 +++++++++++++++++ 4 files changed, 153 insertions(+), 11 deletions(-) diff --git a/mobile/lib/features/channels/send_message_provider.dart b/mobile/lib/features/channels/send_message_provider.dart index 3659bda4bf0..422c3f2d2e2 100644 --- a/mobile/lib/features/channels/send_message_provider.dart +++ b/mobile/lib/features/channels/send_message_provider.dart @@ -1,16 +1,18 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../../shared/relay/relay.dart'; -import '../channels/channel_management_provider.dart'; import '../profile/user_cache_provider.dart'; import '../profile/user_profile.dart'; +import 'channel_management_provider.dart'; import 'channel_messages_provider.dart'; +import 'channels_provider.dart'; /// Sends messages by signing an event with the user's nsec and publishing it /// over the relay's NIP-42-authenticated WebSocket session. class SendMessage { final SignedEventRelay _signedEventRelay; final Future> Function(String channelId) _fetchMembers; + final bool Function(String channelId) _isDmChannel; final Map Function() _readUserCache; final void Function(String channelId, NostrEvent event) _addLocalMessage; final void Function(String channelId, String eventId) _completeLocalMessage; @@ -20,6 +22,7 @@ class SendMessage { required SignedEventRelay signedEventRelay, required Future> Function(String channelId) fetchMembers, + required bool Function(String channelId) isDmChannel, required Map Function() readUserCache, required void Function(String channelId, NostrEvent event) addLocalMessage, required void Function(String channelId, String eventId) @@ -27,6 +30,7 @@ class SendMessage { required void Function(String channelId, String eventId) removeLocalMessage, }) : _signedEventRelay = signedEventRelay, _fetchMembers = fetchMembers, + _isDmChannel = isDmChannel, _readUserCache = readUserCache, _addLocalMessage = addLocalMessage, _completeLocalMessage = completeLocalMessage, @@ -53,12 +57,22 @@ class SendMessage { mentionPubkeys ?? await _resolveMentions(content, channelId); final authorPubkey = _signedEventRelay.pubkey; + // DM messages address every participant. Agent runtimes use these p-tags + // both to wake the addressed agent and to distinguish an intentional DM + // from an unaddressed channel event. Desktop already follows this + // contract; mobile must do the same. + final addressedPubkeys = [...resolvedMentions]; + if (_isDmChannel(channelId)) { + final members = await _fetchMembers(channelId); + addressedPubkeys.addAll(members.map((member) => member.pubkey)); + } + // Normalize mentions: lowercase, deduplicate, exclude self (matching // the desktop's normalizeMentionPubkeys). final selfLower = authorPubkey?.toLowerCase(); final seenMentions = {?selfLower}; final normalizedMentions = [ - for (final pk in resolvedMentions) + for (final pk in addressedPubkeys) if (seenMentions.add(pk.toLowerCase())) pk, ]; @@ -169,6 +183,12 @@ final sendMessageProvider = Provider((ref) { ), fetchMembers: (channelId) => ref.read(channelMembersProvider(channelId).future), + isDmChannel: (channelId) => + ref + .read(channelsProvider) + .value + ?.any((channel) => channel.id == channelId && channel.isDm) ?? + false, readUserCache: () => ref.read(userCacheProvider), addLocalMessage: (channelId, event) => ref .read(channelMessagesProvider(channelId).notifier) diff --git a/mobile/lib/features/profile/presence_cache_provider.dart b/mobile/lib/features/profile/presence_cache_provider.dart index f735c376c35..d06ae6c5cdc 100644 --- a/mobile/lib/features/profile/presence_cache_provider.dart +++ b/mobile/lib/features/profile/presence_cache_provider.dart @@ -12,8 +12,11 @@ import '../../shared/relay/relay.dart'; /// publish presence purely over WS are fine, and TTL expiry will be handled /// by the relay-side `presence:true` filter extension when that lands. class PresenceCacheNotifier extends Notifier> { + static const _backstopInterval = Duration(seconds: 60); + final Set _tracked = {}; void Function()? _presenceUnsub; + Timer? _backstopTimer; int _subscriptionVersion = 0; @override @@ -23,10 +26,18 @@ class PresenceCacheNotifier extends Notifier> { ref.onDispose(() { _presenceUnsub?.call(); _presenceUnsub = null; + _backstopTimer?.cancel(); + _backstopTimer = null; }); if (sessionState.status == SessionStatus.connected) { _subscribePresenceUpdates(); + _refreshTrackedPresence(); + _backstopTimer?.cancel(); + _backstopTimer = Timer.periodic( + _backstopInterval, + (_) => _refreshTrackedPresence(), + ); } return {}; @@ -34,15 +45,40 @@ class PresenceCacheNotifier extends Notifier> { /// Track presence for [pubkeys]. /// - /// Currently a no-op for the actual fetch — we rely on live kind:20001 - /// events. The tracked set is still used to filter incoming events so the - /// cache doesn't grow unbounded. void track(List pubkeys) { final normalized = pubkeys.map((pk) => pk.toLowerCase()).toList(); - _tracked.addAll(normalized); - // TODO(presence): once the relay supports a `presence:true` filter - // extension, issue a one-shot fetch here for the latest known state per - // pubkey. Until then, presence is "online whenever they publish". + var changed = false; + for (final pubkey in normalized) { + changed = _tracked.add(pubkey) || changed; + } + if (changed && + ref.read(relaySessionProvider).status == SessionStatus.connected) { + _refreshTrackedPresence(); + } + } + + /// Hydrate the cache from the relay's Redis-backed presence snapshot. + /// + /// Presence events are ephemeral, so a client that opens after an agent's + /// heartbeat cannot rely on the live subscription alone. + Future _refreshTrackedPresence() async { + if (_tracked.isEmpty) return; + try { + final events = await ref + .read(relaySessionProvider.notifier) + .queryRelay([ + NostrFilter( + kinds: const [EventKind.presenceUpdate], + authors: _tracked.toList(), + limit: _tracked.length, + ), + ]); + for (final event in events) { + _handlePresenceEvent(event); + } + } catch (error) { + debugPrint('[PresenceCacheNotifier] presence refresh failed: $error'); + } } /// Subscribe to kind:20001 presence events over WebSocket. @@ -73,7 +109,13 @@ class PresenceCacheNotifier extends Notifier> { } void _handlePresenceEvent(NostrEvent event) { - final pubkey = event.pubkey.toLowerCase(); + // Relay-synthesized snapshots are relay-signed and identify the actual + // subject with a p-tag. Live events are self-signed by their subject. + final taggedSubject = event.tags + .where((tag) => tag.length >= 2 && tag[0] == 'p') + .map((tag) => tag[1]) + .firstOrNull; + final pubkey = (taggedSubject ?? event.pubkey).toLowerCase(); if (!_tracked.contains(pubkey)) return; final status = event.content; if (status != 'online' && status != 'away' && status != 'offline') return; diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 9c167e10b16..e762b22fc34 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -249,7 +249,9 @@ class RelaySessionNotifier extends Notifier { Future publish( NostrEvent event, { Duration timeout = const Duration(seconds: 8), - }) { + }) async { + await _waitUntilConnected(timeout); + final completer = Completer(); final timer = Timer(timeout, () { @@ -272,6 +274,31 @@ class RelaySessionNotifier extends Notifier { return completer.future; } + Future _waitUntilConnected(Duration timeout) async { + if (state.status == SessionStatus.connected) return; + if (_paused) { + throw StateError('Cannot publish while the app is in the background'); + } + + if (state.status == SessionStatus.disconnected) { + final config = ref.read(relayConfigProvider); + unawaited(_connect(config)); + } + + final deadline = DateTime.now().add(timeout); + while (state.status != SessionStatus.connected) { + if (_disposed || _paused) { + throw StateError('Relay connection unavailable'); + } + if (DateTime.now().isAfter(deadline)) { + throw TimeoutException( + 'Relay did not reconnect within $timeout', + ); + } + await Future.delayed(const Duration(milliseconds: 50)); + } + } + /// Send a raw message over the WebSocket without waiting for acknowledgement. /// Used for ephemeral events like typing indicators. void sendRaw(List payload) { diff --git a/mobile/test/features/channels/send_message_provider_test.dart b/mobile/test/features/channels/send_message_provider_test.dart index f91ce87b2b0..10b4f184f0d 100644 --- a/mobile/test/features/channels/send_message_provider_test.dart +++ b/mobile/test/features/channels/send_message_provider_test.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'package:flutter_test/flutter_test.dart'; import 'package:nostr/nostr.dart' as nostr; +import 'package:buzz/features/channels/channel_management_provider.dart'; import 'package:buzz/features/channels/send_message_provider.dart'; import 'package:buzz/shared/relay/relay.dart'; @@ -19,6 +20,7 @@ void main() { nsec: nostr.Keys.generate().nsec, ), fetchMembers: (_) async => const [], + isDmChannel: (_) => false, readUserCache: () => const {}, addLocalMessage: (_, event) => localMessages.add(event), completeLocalMessage: (_, eventId) => completedIds.add(eventId), @@ -52,6 +54,7 @@ void main() { nsec: nostr.Keys.generate().nsec, ), fetchMembers: (_) async => const [], + isDmChannel: (_) => false, readUserCache: () => const {}, addLocalMessage: (_, event) => localMessages.add(event), completeLocalMessage: (_, eventId) => completedIds.add(eventId), @@ -66,6 +69,56 @@ void main() { expect(completedIds, isEmpty); expect(removedIds, [localMessages.single.id]); }); + + test( + 'addresses every DM participant so agent subscribers receive it', + () async { + final session = _PendingPublishRelaySession(); + final signer = nostr.Keys.generate(); + final agent = nostr.Keys.generate(); + final other = nostr.Keys.generate(); + final send = SendMessage( + signedEventRelay: SignedEventRelay( + session: session, + nsec: signer.nsec, + ), + fetchMembers: (_) async => [ + ChannelMember( + pubkey: signer.public, + role: 'member', + joinedAt: DateTime.fromMillisecondsSinceEpoch(0), + ), + ChannelMember( + pubkey: agent.public, + role: 'bot', + joinedAt: DateTime.fromMillisecondsSinceEpoch(0), + ), + ChannelMember( + pubkey: other.public, + role: 'member', + joinedAt: DateTime.fromMillisecondsSinceEpoch(0), + ), + ], + isDmChannel: (_) => true, + readUserCache: () => const {}, + addLocalMessage: (_, _) {}, + completeLocalMessage: (_, _) {}, + removeLocalMessage: (_, _) {}, + ); + + final result = send(channelId: _channelId, content: 'hello'); + await session.published; + + expect(session.event.tags, containsAll([ + ['p', agent.public], + ['p', other.public], + ])); + expect(session.event.tags, isNot(contains(['p', signer.public]))); + + session.accept(); + await result; + }, + ); } const _channelId = '11111111-1111-4111-8111-111111111111'; From b49115eea28535bb6bad4da6fddb221c713036b3 Mon Sep 17 00:00:00 2001 From: Tryn McCaffery Date: Mon, 3 Aug 2026 13:13:19 -0400 Subject: [PATCH 2/5] fix(acp): retry transient relay query failures Signed-off-by: Tryn McCaffery --- crates/buzz-acp/src/relay.rs | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index c8312cc61e5..7fe2eccf6ce 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -240,7 +240,8 @@ pub struct RestClient { /// Whether an HTTP status code is retriable (transient server/rate-limit errors). fn is_retriable_status(status: reqwest::StatusCode) -> bool { - matches!(status.as_u16(), 429 | 502 | 503 | 504) + let status = status.as_u16(); + status == 408 || status == 429 || (500..600).contains(&status) } /// Base retry delays for transient HTTP failures: 500ms, 1s, 2s. @@ -308,7 +309,7 @@ impl RestClient { } /// Retry helper: executes `build_request` up to 4 times (1 attempt + 3 retries) - /// on transient failures (429, 502, 503, 504, timeout, connect errors). + /// on transient failures (408, 429, 5xx, timeout, connect errors). /// /// NIP-98 auth events are re-signed on each attempt (they have a ±60s window). async fn request_with_retry( @@ -3995,6 +3996,23 @@ async fn wait_for_any_ok( mod tests { use super::*; + #[test] + fn rest_retry_statuses_cover_transient_relay_failures() { + for status in [408, 429, 500, 501, 502, 503, 504, 599] { + assert!( + is_retriable_status(reqwest::StatusCode::from_u16(status).unwrap()), + "HTTP {status} should be retried" + ); + } + + for status in [400, 401, 403, 404, 409, 422] { + assert!( + !is_retriable_status(reqwest::StatusCode::from_u16(status).unwrap()), + "HTTP {status} should fail without retry" + ); + } + } + #[test] fn relay_ws_to_http_plain() { assert_eq!( From bcb93f5c2244960ad9f8b757c6dd86b5e624e698 Mon Sep 17 00:00:00 2001 From: Tryn McCaffery Date: Mon, 3 Aug 2026 16:41:06 -0400 Subject: [PATCH 3/5] fix(acp): survive transient discovery overload Signed-off-by: Tryn McCaffery --- crates/buzz-acp/src/lib.rs | 34 ++++++-- crates/buzz-acp/src/relay.rs | 152 ++++++++++++++++++++++++++++++----- 2 files changed, 163 insertions(+), 23 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index d63f720c651..4a3011222b2 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1351,7 +1351,12 @@ async fn tokio_main() -> Result<()> { tracing::warn!("failed to set startup watermark: {e}"); } - tracing::info!("connected to relay at {}", config.relay_url); + tracing::info!( + target: "buzz_acp::runtime_lifecycle", + runtime_state = "transport_authenticated", + relay_url = %config.relay_url, + "relay transport authenticated; runtime is not ready until discovery and subscriptions finish" + ); relay .subscribe_membership_notifications() @@ -1426,10 +1431,21 @@ async fn tokio_main() -> Result<()> { } } - let channel_info_map = relay - .discover_channels() - .await - .map_err(|e| anyhow::anyhow!("channel discovery error: {e}"))?; + tracing::info!( + target: "buzz_acp::runtime_lifecycle", + runtime_state = "discovering_channels", + "starting canonical channel discovery" + ); + let channel_info_map = relay.discover_channels().await.map_err(|e| { + tracing::error!( + target: "buzz_acp::runtime_lifecycle", + runtime_state = "startup_failed", + startup_stage = "channel_discovery", + error = %e, + "runtime did not become ready because channel discovery failed" + ); + anyhow::anyhow!("channel discovery error: {e}") + })?; tracing::info!("discovered {} channel(s)", channel_info_map.len()); let channel_ids: Vec = channel_info_map.keys().copied().collect(); @@ -1485,6 +1501,14 @@ async fn tokio_main() -> Result<()> { } } + tracing::info!( + target: "buzz_acp::runtime_lifecycle", + runtime_state = "subscriptions_enqueued", + discovered_channels = channel_info_map.len(), + subscribed_channels = subscribed_channel_ids.len(), + "canonical subscriptions are enqueued; runtime may now publish presence" + ); + if let Some((observer, publisher, keys, agent_pubkey, owner_pubkey, owner)) = relay_observer_publisher.take() { diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 7fe2eccf6ce..84b4fe37eea 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -244,12 +244,21 @@ fn is_retriable_status(status: reqwest::StatusCode) -> bool { status == 408 || status == 429 || (500..600).contains(&status) } -/// Base retry delays for transient HTTP failures: 500ms, 1s, 2s. +/// Base retry delays for transient HTTP failures. +/// +/// Managed desktops commonly launch several agents at once. Each agent performs +/// two discovery queries, so a relay recovering from startup can remain +/// overloaded for longer than the old 3.5-second window. Seven total attempts +/// spread that burst over roughly 31.5 seconds (plus jitter) without retrying +/// deterministic auth or request errors. /// Jitter (±20%) is applied at call time via `jittered_duration`. -const REST_RETRY_BASE_DELAYS: [Duration; 3] = [ +const REST_RETRY_BASE_DELAYS: [Duration; 6] = [ Duration::from_millis(500), Duration::from_millis(1000), Duration::from_millis(2000), + Duration::from_millis(4000), + Duration::from_millis(8000), + Duration::from_millis(16000), ]; fn unix_now_secs() -> u64 { @@ -308,7 +317,8 @@ impl RestClient { Ok(format!("Nostr {}", self.sign_nip98(method, url, body)?)) } - /// Retry helper: executes `build_request` up to 4 times (1 attempt + 3 retries) + /// Retry helper: executes `build_request` up to seven times + /// (one attempt plus six retries) /// on transient failures (408, 429, 5xx, timeout, connect errors). /// /// NIP-98 auth events are re-signed on each attempt (they have a ±60s window). @@ -324,24 +334,53 @@ impl RestClient { { let mut last_err = None; + let total_attempts = REST_RETRY_BASE_DELAYS.len() + 1; for (attempt, delay) in std::iter::once(None) .chain(REST_RETRY_BASE_DELAYS.iter().map(|d| Some(*d))) .enumerate() { if let Some(base) = delay { let jittered = jittered_duration(base); - tracing::debug!( - "retrying {method} {path} (attempt {attempt}) in {:.1}s", - jittered.as_secs_f64() + tracing::warn!( + target: "buzz_acp::relay_lifecycle", + relay_state = "http_retry_wait", + method, + path, + next_attempt = attempt + 1, + total_attempts, + delay_seconds = jittered.as_secs_f64(), + "transient relay request failure; waiting before retry" ); tokio::time::sleep(jittered).await; } match build_request().await { - Ok(resp) if resp.status().is_success() => return Ok(resp), + Ok(resp) if resp.status().is_success() => { + if attempt > 0 { + tracing::info!( + target: "buzz_acp::relay_lifecycle", + relay_state = "http_recovered", + method, + path, + attempt = attempt + 1, + total_attempts, + "relay request recovered after transient failures" + ); + } + return Ok(resp); + } Ok(resp) if is_retriable_status(resp.status()) => { let status = resp.status(); - tracing::warn!("{method} {path} returned retriable HTTP {status}"); + tracing::warn!( + target: "buzz_acp::relay_lifecycle", + relay_state = "http_transient_failure", + method, + path, + attempt = attempt + 1, + total_attempts, + http_status = status.as_u16(), + "relay request returned a transient HTTP status" + ); last_err = Some(RelayError::Http(format!( "{method} {path} returned HTTP {status}" ))); @@ -354,15 +393,34 @@ impl RestClient { ))); } Err(e) if e.is_timeout() || e.is_connect() => { - tracing::warn!("{method} {path} network error: {e}"); + tracing::warn!( + target: "buzz_acp::relay_lifecycle", + relay_state = "http_transient_failure", + method, + path, + attempt = attempt + 1, + total_attempts, + error = %e, + "relay request hit a transient network error" + ); last_err = Some(RelayError::Http(e.to_string())); } Err(e) => return Err(RelayError::Http(e.to_string())), } } - Err(last_err - .unwrap_or_else(|| RelayError::Http(format!("{method} {path} failed after retries")))) + let error = last_err + .unwrap_or_else(|| RelayError::Http(format!("{method} {path} failed after retries"))); + tracing::error!( + target: "buzz_acp::relay_lifecycle", + relay_state = "http_retry_exhausted", + method, + path, + total_attempts, + error = %error, + "relay request exhausted its transient retry budget" + ); + Err(error) } /// POST with NIP-98 auth and retry. Re-signs on each attempt. @@ -2918,14 +2976,25 @@ async fn try_autonomous_reconnect( let mut attempt = 0usize; while attempt < backoffs.len() { info!( - "autonomous reconnect attempt {}/{} to {relay_url}…", - attempt + 1, - backoffs.len() + target: "buzz_acp::relay_lifecycle", + relay_state = "reconnecting", + reconnect_mode = "autonomous", + attempt = attempt + 1, + total_attempts = backoffs.len(), + relay_url, + "attempting relay reconnect" ); match do_connect(relay_url, keys, auth_tag).await { Ok((new_ws, handshake_buffer)) => { *ws = new_ws; - info!("autonomous reconnect succeeded (attempt {})", attempt + 1); + info!( + target: "buzz_acp::relay_lifecycle", + relay_state = "transport_reconnected", + reconnect_mode = "autonomous", + attempt = attempt + 1, + relay_url, + "relay transport reconnected; restoring subscriptions" + ); let handshake_ok = process_handshake_buffer( ws, handshake_buffer, @@ -2950,7 +3019,16 @@ async fn try_autonomous_reconnect( match resubscribe_after_reconnect(ws, cmd_rx, state, agent_pubkey_hex, true) .await { - ResubscribeResult::Ok => return ReconnectOutcome::Ok, + ResubscribeResult::Ok => { + info!( + target: "buzz_acp::relay_lifecycle", + relay_state = "online", + reconnect_mode = "autonomous", + active_subscriptions = state.active_subscriptions.len(), + "relay subscriptions restored; event replay is active" + ); + return ReconnectOutcome::Ok; + } ResubscribeResult::Shutdown => return ReconnectOutcome::Shutdown, ResubscribeResult::RetryConnection => { warn!("resubscribe failed after autonomous reconnect — treating as failed attempt"); @@ -3008,6 +3086,13 @@ async fn try_autonomous_reconnect( attempt += 1; } + tracing::error!( + target: "buzz_acp::relay_lifecycle", + relay_state = "reconnect_budget_exhausted", + reconnect_mode = "autonomous", + relay_url, + "bounded autonomous reconnect exhausted; entering persistent reconnect loop" + ); ReconnectOutcome::Failed } @@ -3059,11 +3144,25 @@ async fn wait_for_reconnect( ]; let mut attempt = state.backoff_step; loop { - info!("attempting relay reconnect to {relay_url}…"); + info!( + target: "buzz_acp::relay_lifecycle", + relay_state = "reconnecting", + reconnect_mode = "persistent", + attempt = attempt + 1, + relay_url, + "attempting relay reconnect" + ); match do_connect(relay_url, keys, auth_tag).await { Ok((new_ws, handshake_buffer)) => { *ws = new_ws; - info!("relay reconnected to {relay_url}"); + info!( + target: "buzz_acp::relay_lifecycle", + relay_state = "transport_reconnected", + reconnect_mode = "persistent", + attempt = attempt + 1, + relay_url, + "relay transport reconnected; restoring subscriptions" + ); let handshake_ok = process_handshake_buffer( ws, handshake_buffer, @@ -3086,6 +3185,13 @@ async fn wait_for_reconnect( .await { ResubscribeResult::Ok => { + info!( + target: "buzz_acp::relay_lifecycle", + relay_state = "online", + reconnect_mode = "persistent", + active_subscriptions = state.active_subscriptions.len(), + "relay subscriptions restored; event replay is active" + ); // Drain any commands that arrived during do_connect() + // resubscribe (which don't poll cmd_rx). return drain_post_reconnect(ws, cmd_rx, state, agent_pubkey_hex).await; @@ -4013,6 +4119,16 @@ mod tests { } } + #[test] + fn rest_retry_budget_outlasts_short_relay_startup_overload() { + assert_eq!(REST_RETRY_BASE_DELAYS.len() + 1, 7); + assert_eq!( + REST_RETRY_BASE_DELAYS.iter().sum::(), + Duration::from_millis(31_500), + "managed-agent discovery must not exit after the old 3.5-second window" + ); + } + #[test] fn relay_ws_to_http_plain() { assert_eq!( From 07da14839eb3276413cfc9984155b77d66e97e8f Mon Sep 17 00:00:00 2001 From: Tryn McCaffery Date: Mon, 3 Aug 2026 16:44:39 -0400 Subject: [PATCH 4/5] fix(mobile): reconnect stale relay sessions Signed-off-by: Tryn McCaffery --- mobile/lib/features/channels/compose_bar.dart | 15 +++ .../shared/relay/app_lifecycle_provider.dart | 2 +- mobile/lib/shared/relay/relay_session.dart | 35 ++++-- .../features/channels/compose_bar_test.dart | 31 +++++ .../test/shared/relay/relay_session_test.dart | 107 ++++++++++++++++++ 5 files changed, 181 insertions(+), 9 deletions(-) diff --git a/mobile/lib/features/channels/compose_bar.dart b/mobile/lib/features/channels/compose_bar.dart index a2421c9339b..4f8aab47666 100644 --- a/mobile/lib/features/channels/compose_bar.dart +++ b/mobile/lib/features/channels/compose_bar.dart @@ -495,6 +495,7 @@ class ComposeBar extends HookConsumerWidget { ); isSending.value = true; + uploadError.value = null; try { if (nonMemberAgentPubkeys.isNotEmpty) { await ref @@ -518,6 +519,13 @@ class ComposeBar extends HookConsumerWidget { if (context.mounted) { clearComposer(); } + } catch (error) { + // Keep the draft intact and make a missing relay acknowledgement + // visible. An optimistic composer clear is not proof that the relay + // accepted the message. + if (context.mounted) { + uploadError.value = _formatSendError(error); + } } finally { if (context.mounted) isSending.value = false; } @@ -974,3 +982,10 @@ class ComposeBar extends HookConsumerWidget { ); } } + +String _formatSendError(Object error) { + if (error is TimeoutException) { + return 'Message wasn\u2019t sent because Buzz did not confirm it. Try again.'; + } + return 'Message wasn\u2019t sent. Check your connection and try again.'; +} diff --git a/mobile/lib/shared/relay/app_lifecycle_provider.dart b/mobile/lib/shared/relay/app_lifecycle_provider.dart index 18e3eaebdf6..ac97a915e2c 100644 --- a/mobile/lib/shared/relay/app_lifecycle_provider.dart +++ b/mobile/lib/shared/relay/app_lifecycle_provider.dart @@ -20,7 +20,7 @@ class AppLifecycleNotifier extends Notifier { _connectivitySub = Connectivity().onConnectivityChanged.listen((results) { final hasNetwork = results.any((r) => r != ConnectivityResult.none); if (hasNetwork && state == AppLifecycleState.resumed) { - ref.read(relaySessionProvider.notifier).onAppResumed(); + ref.read(relaySessionProvider.notifier).onNetworkRestored(); } }); diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index e762b22fc34..66ffe5c195e 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -103,6 +103,7 @@ class RelaySessionNotifier extends Notifier { int _subIdCounter = 0; bool _disposed = false; bool _paused = false; + bool _resumeReconnectRequired = false; bool _hasConnectedOnce = false; int _connectionGeneration = 0; @@ -291,9 +292,7 @@ class RelaySessionNotifier extends Notifier { throw StateError('Relay connection unavailable'); } if (DateTime.now().isAfter(deadline)) { - throw TimeoutException( - 'Relay did not reconnect within $timeout', - ); + throw TimeoutException('Relay did not reconnect within $timeout'); } await Future.delayed(const Duration(milliseconds: 50)); } @@ -342,6 +341,10 @@ class RelaySessionNotifier extends Notifier { /// Called by the app lifecycle provider when the app goes to background. void onAppPaused() { + // Do not trust an iOS socket after any background transition. The process + // may be suspended before the grace timer runs, leaving Dart's local + // socket state as `connected` even though the network path is gone. + _resumeReconnectRequired = true; _backgroundGraceTimer?.cancel(); _backgroundGraceTimer = Timer(const Duration(seconds: 5), _pauseNow); } @@ -361,12 +364,28 @@ class RelaySessionNotifier extends Notifier { _backgroundGraceTimer?.cancel(); _backgroundGraceTimer = null; - // If still connected, nothing to do — the socket survived the background - // grace window. - if (state.status == SessionStatus.connected) return; + if (!_resumeReconnectRequired && state.status == SessionStatus.connected) { + return; + } + _resumeReconnectRequired = false; - // Cancel any in-flight reconnect backoff timer so we reconnect immediately - // instead of waiting for the (possibly large) exponential delay. + // Reconnect even when the socket still claims to be connected. On iOS a + // backgrounded or path-migrated socket can remain locally open while no + // longer receiving relay events; replacing it is what makes the first + // message after foregrounding observable. + _reconnectTimer?.cancel(); + _reconnectDelayMs = _baseReconnectDelayMs; + final config = ref.read(relayConfigProvider); + _connect(config); + } + + /// Replace the socket immediately after the network becomes reachable. + /// + /// Connectivity restoration is stronger evidence than the websocket's + /// local state: a half-open socket may still report connected after Wi-Fi, + /// cellular, VPN, or Tailnet path migration. + void onNetworkRestored() { + if (_disposed || _paused) return; _reconnectTimer?.cancel(); _reconnectDelayMs = _baseReconnectDelayMs; final config = ref.read(relayConfigProvider); diff --git a/mobile/test/features/channels/compose_bar_test.dart b/mobile/test/features/channels/compose_bar_test.dart index bf236e8151d..25a1d4179d2 100644 --- a/mobile/test/features/channels/compose_bar_test.dart +++ b/mobile/test/features/channels/compose_bar_test.dart @@ -393,6 +393,37 @@ void main() { }); group('ComposeBar', () { + testWidgets('failed send keeps the draft and shows a truthful error', ( + tester, + ) async { + await tester.pumpWidget( + _buildComposeBar( + uploadService: _testUploadService(nostr.Keys.generate().nsec), + onSend: + ( + content, + mentionPubkeys, { + mediaTags = const >[], + }) async { + throw TimeoutException('relay acknowledgement missing'); + }, + ), + ); + + await _expandComposer(tester); + await tester.enterText(find.byType(TextField), 'please persist'); + await tester.tap(find.byIcon(LucideIcons.arrowUp)); + await tester.pump(); + + expect(find.text('please persist'), findsOneWidget); + expect( + find.text( + 'Message wasn\u2019t sent because Buzz did not confirm it. Try again.', + ), + findsOneWidget, + ); + }); + testWidgets('mounted composer does not carry draft text across an in-place ' 'identity switch', (tester) async { final keysA = nostr.Keys.generate(); diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 6cd601f56b0..f6b68273e31 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -260,6 +260,113 @@ void main() { expect(session.state.status, SessionStatus.disconnected); }); + test( + 'foreground replaces a socket that still claims to be connected', + () async { + final sockets = <_ControlledRelaySocket>[]; + final keychain = nostr.Keys.generate(); + final session = RelaySessionNotifier( + socketFactory: + ({ + required wsUrl, + required nsec, + required onMessage, + required onConnected, + required onDisconnected, + }) { + final socket = _ControlledRelaySocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + ); + sockets.add(socket); + return socket; + }, + ); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => session), + relayConfigProvider.overrideWith( + () => _FakeRelayConfigNotifier( + baseUrl: 'https://relay.example', + nsec: keychain.nsec, + ), + ), + authProvider.overrideWith(() => _AuthenticatedAuthNotifier()), + ], + ); + addTearDown(container.dispose); + await container.read(authProvider.future); + final subscription = container.listen(relaySessionProvider, (_, _) {}); + addTearDown(subscription.close); + await Future.delayed(Duration.zero); + + sockets.single.connectSuccessfully(); + expect(session.state.status, SessionStatus.connected); + + session.onAppPaused(); + session.onAppResumed(); + await Future.delayed(Duration.zero); + + expect(sockets, hasLength(2)); + expect(session.state.status, SessionStatus.reconnecting); + sockets.last.connectSuccessfully(); + expect(session.state.status, SessionStatus.connected); + }, + ); + + test('network restoration replaces a half-open connected socket', () async { + final sockets = <_ControlledRelaySocket>[]; + final keychain = nostr.Keys.generate(); + final session = RelaySessionNotifier( + socketFactory: + ({ + required wsUrl, + required nsec, + required onMessage, + required onConnected, + required onDisconnected, + }) { + final socket = _ControlledRelaySocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + ); + sockets.add(socket); + return socket; + }, + ); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => session), + relayConfigProvider.overrideWith( + () => _FakeRelayConfigNotifier( + baseUrl: 'https://relay.example', + nsec: keychain.nsec, + ), + ), + authProvider.overrideWith(() => _AuthenticatedAuthNotifier()), + ], + ); + addTearDown(container.dispose); + await container.read(authProvider.future); + final subscription = container.listen(relaySessionProvider, (_, _) {}); + addTearDown(subscription.close); + await Future.delayed(Duration.zero); + + sockets.single.connectSuccessfully(); + session.onNetworkRestored(); + await Future.delayed(Duration.zero); + + expect(sockets, hasLength(2)); + sockets.last.connectSuccessfully(); + expect(session.state.status, SessionStatus.connected); + }); + test('delivers the same live event to each matching subscription', () async { final session = RelaySessionNotifier(); final firstEvents = []; From d8b387499cdded29faa2781979f983ec33a2b5f6 Mon Sep 17 00:00:00 2001 From: Tryn McCaffery Date: Mon, 3 Aug 2026 16:47:16 -0400 Subject: [PATCH 5/5] fix(desktop): surface relay publish failures Signed-off-by: Tryn McCaffery --- .../messages/ui/useMentionSendFlow.ts | 10 ++++++++- desktop/tests/e2e/channels.spec.ts | 22 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/desktop/src/features/messages/ui/useMentionSendFlow.ts b/desktop/src/features/messages/ui/useMentionSendFlow.ts index 6d4e007cd4f..45e3bd98729 100644 --- a/desktop/src/features/messages/ui/useMentionSendFlow.ts +++ b/desktop/src/features/messages/ui/useMentionSendFlow.ts @@ -542,7 +542,7 @@ export function useMentionSendFlow({ [...draft.savedSpoileredAttachmentUrls], ); } - } catch { + } catch (error) { // Only restore the composer content if the user is still on the // channel that originated the send. if (draft.capturedChannelId === channelIdRef.current) { @@ -554,6 +554,14 @@ export function useMentionSendFlow({ new Set(draft.savedSpoileredAttachmentUrls), ); } + + // A relay rejection or publish timeout must never look like a + // successful send. Keep a visible, durable-in-view receipt alongside + // the restored draft so the user can retry deliberately. + const message = + error instanceof Error ? error.message : "Failed to send message."; + setNonMemberPromptError(message); + toast.error(message); } } finally { isCompleteSendPendingRef.current = false; diff --git a/desktop/tests/e2e/channels.spec.ts b/desktop/tests/e2e/channels.spec.ts index 9da40224205..1b192626335 100644 --- a/desktop/tests/e2e/channels.spec.ts +++ b/desktop/tests/e2e/channels.spec.ts @@ -972,6 +972,28 @@ test("does not reroute an expanded DM after the channel pane unmounts", async ({ .toBeNull(); }); +test("shows a relay failure and restores the draft in an existing DM", async ({ + page, +}) => { + const sendError = "Mock existing DM publish failed."; + const message = "@Fizz please inspect this"; + await installMockBridge(page, { + sendMessageErrors: [sendError], + }); + await page.goto("/"); + await page.getByTestId("channel-alice-tyler").click(); + + const input = page.getByTestId("message-input"); + await input.fill(message); + await page.getByTestId("send-message").click(); + + await expect(page.getByText(sendError).first()).toBeVisible(); + await expect(input).toContainText(message); + await expect( + page.getByTestId("message-timeline").getByText(message, { exact: true }), + ).toHaveCount(0); +}); + test("drops an expanded DM after the first message fails", async ({ page }) => { const retryMessage = "Retry without the agent"; const sendError = "Mock first DM send failed.";