External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(mobile): retry channel-sections startup sync when relay rate-limits cold start (#3004)

Browse Source 
**Category:** fix
**User Impact:** Channel groups created on desktop now reliably appear
on Android and iOS on cold start, instead of falling back to the default
ungrouped list.

**Problem:** On mobile cold start, ChannelsNotifier fires ~25
per-channel REQs at once, exhausting the relay's per-connection
rate-limit quota. `ChannelSectionsManager` then gets BOTH its one-shot
history fetch and its live subscription rejected with `rate-limited:
quota exceeded` — and both errors were silently swallowed (`catch (_)`)
with no retry, so the manager kept the local (empty/default) store
forever. Restarting the app repeats the same storm, so Android reliably
lost the race every launch. Captured live on the emulator with
instrumentation.

**Solution:** Track whether the startup fetch and the live subscription
have each succeeded, and retry `_syncWithRelay` with exponential backoff
(2s base, shift-capped, 30s max) until both land. The retry timer is
cancelled on dispose, and previously-swallowed errors are now logged.

Based directly on `main` — independent of #2829 (which fixes the *write*
path: unpublished local edits being clobbered). The analogous retry for
`ChannelSortManager` lives in #2829, since that manager is introduced
there.

<details>
<summary>File changes</summary>


**mobile/lib/features/channels/channel_sections/channel_sections_manager.dart**
Extract the startup fetch + live-subscription into `_syncWithRelay`,
track success of each step, and schedule a backoff retry until both
succeed. `_fetchAndMerge` and `_startLiveSubscription` now report
success; swallowed errors are logged; retry timer cancelled on dispose.
`startupRetryBaseDelay` ctor param is test-visible.


**mobile/test/features/channels/channel_sections/channel_sections_manager_test.dart**
New regression tests with a rate-limiting relay fake: remote sections
are adopted after retries; retry stops once fetch + subscription
succeed; dispose cancels pending retries.

</details>

## Reproduction Steps

1. On desktop, create channel groups (sections) for an account.
2. Cold-start the Android app for the same account on a relay with
per-connection rate limiting and enough joined channels to trigger the
REQ burst (~25 channels reproduced it reliably).
3. Before this fix: logs show `fetch FAILED: Exception: rate-limited:
quota exceeded` and the live subscription failing, then silence — the
channel list renders the default ungrouped list forever, surviving app
restarts.
4. With this fix: logs show `startup sync incomplete; retrying in 2000ms
(attempt 1)`, the retry succeeds, and the desktop-created groups render.

## Verification

- Live on emulator-5554 (earlier stacked build of the same logic): cold
start reproduced the manager being rate-limited, then a single 2s retry
succeeding and groups rendering, matching desktop channel-for-channel.
- Full mobile suite run at this exact head (c5f1d9a38): 654 passing; the
4 failures (3× `compose_bar_test`, 1× `channels_page_test`) reproduce on
unmodified `main` (74b63e184) — pre-existing, unrelated. `flutter
analyze` clean on both touched files.

Originating thread: Buzz channel ed3994af-0949-447c-be00-29f03965b52e,
root 4bf7cbfd48bf.

---------

Signed-off-by: npub1223z34hd7vtwc6qj4s7flsxkj644nlre2nthu7lrrmkumhu3xddsrx9r6w <52a228d6edf316ec6812ac3c9fc0d696ab59fc7954d77e7be31eedcddf91335b@buzz.block.builderlab.xyz>
Co-authored-by: npub1223z34hd7vtwc6qj4s7flsxkj644nlre2nthu7lrrmkumhu3xddsrx9r6w <52a228d6edf316ec6812ac3c9fc0d696ab59fc7954d77e7be31eedcddf91335b@buzz.block.builderlab.xyz>
This commit is contained in:
Taylor Ho
2026-07-27 10:43:21 -07:00
committed by GitHub
co-authored by npub1223z34hd7vtwc6qj4s7flsxkj644nlre2nthu7lrrmkumhu3xddsrx9r6w
parent b92a1f4bf4
commit e28707f6b2
2 changed files with 413 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mobile/lib/features/channels/channel_sections/channel_sections_manager.dart
+101 -9
View File
@@ -48,6 +48,12 @@ class ChannelSectionsManager {
void Function()? _unsubscribe;
bool _disposed = false;
/// Base delay for the startup-sync retry backoff. Overridable in tests.
final Duration _startupRetryBaseDelay;
Timer? _startupRetryTimer;
int _startupRetryAttempt = 0;
bool _startupFetchSucceeded = false;
ChannelSectionsManager({
required this.pubkey,
required SharedPreferences prefs,
@@ -56,12 +62,15 @@ class ChannelSectionsManager {
required SignedEventRelay? signedEventRelay,
required bool remoteEnabled,
required VoidCallback onChanged,
@visibleForTesting
Duration startupRetryBaseDelay = const Duration(seconds: 2),
}) : _storage = ChannelSectionsStorage(prefs),
_crypto = crypto,
_relaySession = relaySession,
_signedEventRelay = signedEventRelay,
_remoteEnabled = remoteEnabled,
_onChanged = onChanged,
_startupRetryBaseDelay = startupRetryBaseDelay,
_store = ChannelSectionsStorage(prefs).read(pubkey);
ChannelSectionStore get store => _store;
@@ -74,15 +83,73 @@ class ChannelSectionsManager {
return;
}
await _fetchAndMerge();
await _startLiveSubscription();
await _syncWithRelay();
_onChanged();
}
/// One startup-sync attempt: fetch the remote blob, then start the live
/// subscription. Either step can lose a transient race on cold start (the
/// relay rate-limits the burst of per-channel subscriptions and rejects
/// with `rate-limited: quota exceeded`) — retry with backoff instead of
/// silently giving up, which left desktop-created groups invisible until
/// an unrelated refetch.
///
/// Retries are intentionally unbounded for the manager's lifetime: this
/// sync must eventually land for groups to appear at all, and at the 30s
/// delay ceiling a persistent retry is cheap. Do not "fix" this into a
/// bounded loop — giving up permanently is the exact bug this replaces.
Future<void> _syncWithRelay() async {
if (!_startupFetchSucceeded) {
_startupFetchSucceeded = await _fetchAndMerge();
}
final subscribed = _unsubscribe != null || await _startLiveSubscription();
if (!_startupFetchSucceeded || !subscribed) {
_scheduleStartupRetry();
} else {
// Fully recovered — later transient failures (e.g. a late relay
// CLOSED) start backing off from the base delay again instead of the
// ceiling the cold start climbed to.
_startupRetryAttempt = 0;
}
}
void _scheduleStartupRetry() {
if (_disposed) return;
_startupRetryTimer?.cancel();
// The inner shift cap is overflow protection, not the delay policy: the
// consecutive-failure counter is unbounded, and an unchecked `<<`
// past 62 wraps negative, which would make the Timer fire immediately in
// a hot loop. At the default 2s base the outer 30s clamp is what callers
// actually observe (2s, 4s, …, 30s); the shift cap only bites for the
// tiny injected bases used in tests.
final delayMs = min(
_startupRetryBaseDelay.inMilliseconds << min(_startupRetryAttempt, 5),
30000,
);
_startupRetryAttempt++;
debugPrint(
'[ChannelSectionsManager] startup sync incomplete; '
'retrying in ${delayMs}ms (attempt $_startupRetryAttempt)',
);
_startupRetryTimer = Timer(Duration(milliseconds: delayMs), () {
_startupRetryTimer = null;
unawaited(
_syncWithRelay().then((_) {
if (!_disposed) _onChanged();
}),
);
});
}
void dispose({bool flushPending = true}) {
if (_disposed) return;
_disposed = true;
_startupRetryTimer?.cancel();
_startupRetryTimer = null;
final hadPending = _publishDebounce != null;
_publishDebounce?.cancel();
_publishDebounce = null;
@@ -201,8 +268,10 @@ class ChannelSectionsManager {
});
}
Future<void> _fetchAndMerge() async {
if (_relaySession == null) return;
/// Returns whether the fetch reached the relay (regardless of whether a
/// remote blob exists).
Future<bool> _fetchAndMerge() async {
if (_relaySession == null) return false;
try {
final events = await _relaySession.fetchHistory(
NostrFilter(
@@ -217,13 +286,17 @@ class ChannelSectionsManager {
_mergeEvents(events);
_persist();
if (!_disposed) _onChanged();
} catch (_) {
return true;
} catch (error) {
debugPrint('[ChannelSectionsManager] fetch failed: $error');
// Local state remains usable when relay is unavailable.
return false;
}
}
Future<void> _startLiveSubscription() async {
if (_relaySession == null) return;
/// Returns whether the live subscription was established.
Future<bool> _startLiveSubscription() async {
if (_relaySession == null) return false;
try {
_unsubscribe = await _relaySession.subscribe(
NostrFilter(
@@ -235,12 +308,31 @@ class ChannelSectionsManager {
limit: 1,
),
_handleIncomingEvent,
onClosed: _handleSubscriptionClosed,
);
} catch (_) {
// Non-fatal — local state and history still work.
return true;
} catch (error) {
debugPrint('[ChannelSectionsManager] live subscription failed: $error');
// Non-fatal — local state and history still work; retried by the
// startup-sync backoff.
return false;
}
}
/// A relay `CLOSED` can arrive after `subscribe()` already reported
/// success: the 500ms readiness wait times out silently under load, and
/// the rate-limit rejection lands later. Without this handler the manager
/// would keep a dead subscription and never retry — the exact
/// load-correlated cold-start failure this retry exists for.
void _handleSubscriptionClosed(String message) {
if (_disposed) return;
debugPrint(
'[ChannelSectionsManager] live subscription closed by relay: $message',
);
_unsubscribe = null;
_scheduleStartupRetry();
}
void _mergeEvents(List<NostrEvent> events) {
for (final event in events) {
if (event.pubkey != pubkey) continue;
mobile/test/features/channels/channel_sections/channel_sections_manager_test.dart
+312
View File
@@ -0,0 +1,312 @@
import 'dart:convert';
import 'package:flutter_test/flutter_test.dart';
import 'package:nostr/nostr.dart' as nostr;
import 'package:shared_preferences/shared_preferences.dart';
import 'package:buzz/features/channels/channel_sections/channel_sections_manager.dart';
import 'package:buzz/shared/relay/relay.dart';
void main() {
late SharedPreferences prefs;
late nostr.Keys keychain;
late ChannelSectionsCrypto crypto;
Future<void> setUpEnv() async {
SharedPreferences.setMockInitialValues({});
prefs = await SharedPreferences.getInstance();
keychain = nostr.Keys.generate();
crypto = ChannelSectionsCrypto(keychain.nsec, keychain.public);
}
NostrEvent sectionsEvent({
required List<Map<String, dynamic>> sections,
Map<String, String> assignments = const {},
required int createdAt,
String id = 'remote-event',
}) {
final payload = jsonEncode({
'version': 1,
'sections': sections,
'assignments': assignments,
});
return NostrEvent(
id: id,
pubkey: keychain.public,
createdAt: createdAt,
kind: EventKind.readState,
tags: const [
['d', 'channel-sections'],
['t', 'channel-sections'],
],
content: crypto.encrypt(payload),
sig: 'sig',
);
}
ChannelSectionsManager buildManager({
required RelaySessionNotifier relaySession,
Duration startupRetryBaseDelay = const Duration(milliseconds: 5),
}) {
return ChannelSectionsManager(
pubkey: keychain.public,
prefs: prefs,
crypto: crypto,
relaySession: relaySession,
signedEventRelay: null,
remoteEnabled: true,
onChanged: () {},
startupRetryBaseDelay: startupRetryBaseDelay,
);
}
test('startup fetch rejected by relay rate limit retries until the remote '
'blob is adopted (cold-start regression)', () async {
await setUpEnv();
// Cold start: the relay rejects the first fetch AND the first live
// subscription with `rate-limited: quota exceeded` because the channel
// list fired dozens of REQs first. Pre-fix, both errors were swallowed
// and desktop-created groups never appeared until app data was cleared.
final relay = _RateLimitedRelaySession(
failuresBeforeSuccess: 2,
historyEvents: [
sectionsEvent(
sections: [
{'id': 's1', 'name': 'Desktop Group', 'order': 0},
],
createdAt: 100,
),
],
);
final manager = buildManager(
relaySession: relay,
startupRetryBaseDelay: const Duration(milliseconds: 10),
);
await manager.initialize();
expect(
manager.store.sections,
isEmpty,
reason: 'first fetch lost the rate-limit race',
);
// Wait for the backoff retries (10ms, 20ms, …) to win the race.
await _waitUntil(() => manager.store.sections.isNotEmpty);
expect(manager.store.sections.single.name, 'Desktop Group');
expect(
relay.subscribeCalls,
greaterThan(1),
reason: 'live subscription must be retried too',
);
manager.dispose(flushPending: false);
});
test(
'startup retry stops after fetch and subscription both succeed',
() async {
await setUpEnv();
final relay = _RateLimitedRelaySession(
failuresBeforeSuccess: 1,
historyEvents: [
sectionsEvent(
sections: [
{'id': 's1', 'name': 'Desktop Group', 'order': 0},
],
createdAt: 100,
),
],
);
final manager = buildManager(relaySession: relay);
await manager.initialize();
await _waitUntil(() => manager.store.sections.isNotEmpty);
final fetchCallsAfterSuccess = relay.fetchCalls;
final subscribeCallsAfterSuccess = relay.subscribeCalls;
await Future<void>.delayed(const Duration(milliseconds: 60));
expect(relay.fetchCalls, fetchCallsAfterSuccess);
expect(relay.subscribeCalls, subscribeCallsAfterSuccess);
manager.dispose(flushPending: false);
},
);
test('disposing the manager cancels pending startup retries', () async {
await setUpEnv();
final relay = _RateLimitedRelaySession(failuresBeforeSuccess: 1000);
final manager = buildManager(relaySession: relay);
await manager.initialize();
manager.dispose(flushPending: false);
final fetchCallsAtDispose = relay.fetchCalls;
await Future<void>.delayed(const Duration(milliseconds: 60));
expect(
relay.fetchCalls,
fetchCallsAtDispose,
reason: 'no retries may fire after dispose',
);
});
test('late relay CLOSED after subscribe() reported success retries and '
'eventually adopts remote state', () async {
await setUpEnv();
// Under cold-start load, subscribe() can time out its 500ms readiness
// wait and resolve successfully before the relay's rate-limit CLOSED
// lands. The rejection then arrives only via onClosed. Pre-fix the
// manager passed no onClosed, kept the dead subscription, and never
// retried.
final relay = _RateLimitedRelaySession(failuresBeforeSuccess: 0);
final manager = buildManager(relaySession: relay);
await manager.initialize();
expect(relay.subscribeCalls, 1);
// Late CLOSED lands after initialize() completed successfully.
relay.closeLiveSubscription('rate-limited: quota exceeded');
// The manager must drop the dead subscription and re-subscribe.
await _waitUntil(() => relay.subscribeCalls > 1);
// Remote state arriving on the NEW subscription must be adopted.
relay.emit(
sectionsEvent(
sections: [
{'id': 's1', 'name': 'Desktop Group', 'order': 0},
],
createdAt: 100,
),
);
await _waitUntil(() => manager.store.sections.isNotEmpty);
expect(manager.store.sections.single.name, 'Desktop Group');
// Exactly one replacement subscription — no duplicates.
await Future<void>.delayed(const Duration(milliseconds: 60));
expect(relay.subscribeCalls, 2);
expect(relay.activeListeners, 1);
manager.dispose(flushPending: false);
});
test('late CLOSED after dispose does not schedule retries', () async {
await setUpEnv();
final relay = _RateLimitedRelaySession(failuresBeforeSuccess: 0);
final manager = buildManager(relaySession: relay);
await manager.initialize();
manager.dispose(flushPending: false);
relay.closeLiveSubscription('rate-limited: quota exceeded');
await Future<void>.delayed(const Duration(milliseconds: 60));
expect(relay.subscribeCalls, 1, reason: 'no re-subscribe after dispose');
});
test('backoff resets after full recovery so later failures start from the '
'base delay', () async {
await setUpEnv();
// Climb the failure counter: 5 rate-limited attempts before success.
// With a 5ms base, attempt 5 would wait 5<<5 = 160ms if the counter
// were never reset.
final relay = _RateLimitedRelaySession(failuresBeforeSuccess: 5);
final manager = buildManager(relaySession: relay);
await manager.initialize();
await _waitUntil(() => relay.subscribeCalls > 5);
// Fully recovered. A later transient failure (late CLOSED) must retry
// from the base delay, not the climbed ceiling.
final subscribeCallsAfterRecovery = relay.subscribeCalls;
relay.closeLiveSubscription('rate-limited: quota exceeded');
final stopwatch = Stopwatch()..start();
await _waitUntil(() => relay.subscribeCalls > subscribeCallsAfterRecovery);
stopwatch.stop();
expect(
stopwatch.elapsedMilliseconds,
lessThan(100),
reason:
'retry after recovery must wait ~base delay (5ms), '
'not the pre-recovery backoff (160ms)',
);
manager.dispose(flushPending: false);
});
}
Future<void> _waitUntil(
bool Function() condition, {
Duration timeout = const Duration(seconds: 2),
}) async {
final deadline = DateTime.now().add(timeout);
while (!condition()) {
if (DateTime.now().isAfter(deadline)) {
fail('condition not met within $timeout');
}
await Future<void>.delayed(const Duration(milliseconds: 10));
}
}
/// Rejects the first [failuresBeforeSuccess] fetch and subscribe calls with
/// the relay's rate-limit error, then succeeds — the exact failure mode seen
/// on Android cold start where the channel-list REQ burst exhausts the
/// relay's per-connection quota before the sections manager gets a turn.
class _RateLimitedRelaySession extends RelaySessionNotifier {
_RateLimitedRelaySession({
required this.failuresBeforeSuccess,
this.historyEvents = const [],
});
final int failuresBeforeSuccess;
final List<NostrEvent> historyEvents;
int fetchCalls = 0;
int subscribeCalls = 0;
final List<void Function(NostrEvent)> _listeners = [];
final List<void Function(String)> _closedCallbacks = [];
int get activeListeners => _listeners.length;
void emit(NostrEvent event) {
for (final listener in List.of(_listeners)) {
listener(event);
}
}
/// Simulates a relay CLOSED landing after subscribe() already resolved —
/// the readiness-timeout window on a loaded cold start. Drops the live
/// subscription (as _handleClosed does) and invokes onClosed.
void closeLiveSubscription(String message) {
if (_listeners.isEmpty) return;
_listeners.removeAt(0);
final onClosed = _closedCallbacks.removeAt(0);
onClosed(message);
}
@override
Future<List<NostrEvent>> fetchHistory(
NostrFilter filter, {
Duration timeout = const Duration(seconds: 8),
}) async {
fetchCalls++;
if (fetchCalls <= failuresBeforeSuccess) {
throw Exception('rate-limited: quota exceeded; retry in 2s');
}
return historyEvents;
}
@override
Future<void Function()> subscribe(
NostrFilter filter,
void Function(NostrEvent) onEvent, {
void Function(String message)? onClosed,
}) async {
subscribeCalls++;
if (subscribeCalls <= failuresBeforeSuccess) {
throw Exception('rate-limited: quota exceeded; retry in 1s');
}
_listeners.add(onEvent);
_closedCallbacks.add(onClosed ?? (_) {});
return () {
final index = _listeners.indexOf(onEvent);
if (index >= 0) {
_listeners.removeAt(index);
_closedCallbacks.removeAt(index);
}
};
}
}