jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-29 21:41:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mobile): keep read state across reconnects and resume (#8672)

Browse Source 
### Context
Every app resume and relay reconnect tore down `ReadStateManager` and
built a new one. `ReadStateNotifier.build()` watched the whole
`relaySessionProvider` state, so each status change (disconnected,
connecting, connected) rebuilt the notifier. On an iPhone 12 mini each
resume re-ran the full read-state fetch and decrypt (about 19 to 24
decrypts, 60 to 95 ms).

The rebuild also cleared `_forcedUnreadContexts`, so a message or
channel the user marked unread lost that flag on any connection change.

The watch dates from #422, when the manager was built with
`remoteEnabled` tied to the connection status. #802 made `remoteEnabled`
constant and added the `ref.listen` that calls `reinitializeRemote()` on
connect, but left the watch behind.

### Change
- Watch `relaySessionProvider.notifier` instead of its state. Config,
nsec, community, pubkey, crypto and a replaced session still rebuild.
Status changes don't.
- Reconnects re-sync through the existing listener
(`reinitializeRemote()`: re-fetch, then re-subscribe, preserving the
history and live overlap).
- Re-syncs are serialized. One runs at a time after any pending
`initialize()`, and reconnects that arrive while one waits share it.
Overlapping passes could otherwise each install a live subscription and
leak one.
- A connect that lands while the first `initialize()` is running (the
cold-start connect) skips the re-sync when that initialize fetched and
subscribed successfully. Without this, cold start did the read-state
work twice.

### Validation
- Provider journey: disconnected, connecting and reconnecting cause no
refetch, no re-subscribe and no readiness drop, and the forced-unread
flag and read mark survive. Connected runs exactly one refetch and one
live-subscription swap. A second reconnect behaves the same. Account and
community switches still rebuild and clear forced flags.
- Manager tests: a connect during a successful `initialize()` does not
sync again; one during a failed `initialize()` does, and leaves one live
subscription; three reconnects during a running re-sync produce one more
shared pass and one live subscription.
- Falsified: restoring the old watch fails the reconnect journey
(`isReady` false, flag lost). The previous version without serialized
re-syncs fails the overlap test (5 passes instead of 3).
- `just mobile-check` and `just mobile-test` pass.

iPhone 12 mini (Profile builds of base and this branch, real community
over the production relay, instrumented manager lifecycle and
live-subscription counts):

| | base | this branch |
| --- | --- | --- |
| Resume after 3 s, 10 s (x3), 60 s background | 3 new managers and 3
initializes per resume | no new manager, 1 re-sync |
| Live read-state subscriptions after resume | 1 | 1 |
| Channel marked unread, then 10 s / 60 s background | flag lost after
the first resume | still unread after both |
| Read-state work per resume (median) | ~24 decrypts, ~176 ms main
isolate | ~18 decrypts, ~125 ms |
| Unread badges | same bold set | same bold set |

The first device run also found the doubled cold-start sync, fixed
above. A re-check on this head:

| Cold launch | base | first version | this head |
| --- | --- | --- | --- |
| Managers built | 3 | 1 | 1 |
| Read-state syncs | 2 initializes | 1 initialize + 1 re-sync | 1
initialize |
| Decrypts | 19 | 30 | 15 |
| Main-isolate ms in read-state code | 124 | 185 to 197 | 94 to 98 |

Four resumes on this head: no new manager, 1 re-sync and 1 live
subscription each, 17 to 20 decrypts. A foreground socket drop was not
exercised on the device; every resume there was a full disconnect and
reconnect.

buzz-review-completed

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Tom Brow <tomb@block.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Tom Brow
2026-10-29 09:12:17 -07:00
committed by GitHub
parent 5ea5ae1727
commit 8abcf2ea2a
4 changed files with 336 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
mobile/lib/shared/read_state/read_state_manager.dart
View File

@@ -64,6 +64,15 @@ class ReadStateManager {
Timer? _debounceTimer;
void Function()? _unsubscribeLive;
bool _initialized = false;
// Pending first fetch and subscribe. A reconnect while it runs waits for it
// so two passes cannot each install a live subscription and leak one.
Future<void>? _initializing;
// The reconnect pass running now, and the one waiting to start after it.
Future<void>? _runningResync;
Future<void>? _queuedResync;
// Whether the first fetch and subscribe both succeeded. A connect that
// lands while they run (the cold-start connect) then has nothing to redo.
bool _initialSyncSucceeded = false;
bool _disposed = false;
bool _isPublishing = false;
// Set when a publish is asked for while one is running. The running
@@ -111,8 +120,16 @@ class ReadStateManager {
return;
}
await _fetchAndMerge();
await _startLiveSubscription();
final initializing = Completer<void>();
_initializing = initializing.future;
try {
final fetched = await _fetchAndMerge();
final subscribed = await _startLiveSubscription();
_initialSyncSucceeded = fetched && subscribed;
} finally {
_initializing = null;
initializing.complete();
}
if (!_isIdenticalToLastPublished(_currentContexts())) {
_schedulePublish();
}
@@ -162,8 +179,43 @@ class ReadStateManager {
await _publish();
}
Future<void> reinitializeRemote() async {
if (_disposed || !_remoteEnabled || !_initialized) return;
/// Fetch and resubscribe after a reconnect. Passes run one at a time, so
/// each replaces the previous live subscription instead of leaking one.
/// Reconnects while a pass waits to start share that pass.
Future<void> reinitializeRemote() {
if (_disposed || !_remoteEnabled || !_initialized) return Future.value();
final queued = _queuedResync;
if (queued != null) return queued;
final running = _runningResync;
final initializing = _initializing;
final pass = Completer<void>();
_queuedResync = pass.future;
unawaited(() async {
try {
if (running != null) {
await running;
} else if (initializing != null) {
await initializing;
// Connected while the first sync ran: it already used this
// connection.
if (_initialSyncSucceeded) return;
}
if (_disposed) return;
if (identical(_queuedResync, pass.future)) _queuedResync = null;
_runningResync = pass.future;
await _resyncRemote();
} catch (e) {
debugPrint('[ReadStateManager] reinitializeRemote failed: $e');
} finally {
if (identical(_queuedResync, pass.future)) _queuedResync = null;
if (identical(_runningResync, pass.future)) _runningResync = null;
pass.complete();
}
}());
return pass.future;
}
Future<void> _resyncRemote() async {
debugPrint('[ReadStateManager] reinitializeRemote');
if (_isPublishing) {
await _publishCompleter?.future;
@@ -221,7 +273,7 @@ class ReadStateManager {
return true;
}
Future<void> _fetchAndMerge() async {
Future<bool> _fetchAndMerge() async {
try {
final events = await _relaySession!.fetchHistory(
NostrFilter(
@@ -237,8 +289,10 @@ class ReadStateManager {
_mergeEvents(events);
_persistLocalState();
_onChanged();
return true;
} catch (e) {
debugPrint('[ReadStateManager] fetchAndMerge failed: $e');
return false;
}
}
@@ -310,7 +364,7 @@ class ReadStateManager {
return changed;
}
Future<void> _startLiveSubscription() async {
Future<bool> _startLiveSubscription() async {
try {
final unsub = await _relaySession!.subscribe(
NostrFilter(
@@ -325,12 +379,14 @@ class ReadStateManager {
);
if (_disposed) {
unsub.call();
return;
return false;
}
_unsubscribeLive = unsub;
debugPrint('[ReadStateManager] live subscription established');
return true;
} catch (e) {
debugPrint('[ReadStateManager] live subscription FAILED: $e');
return false;
}
}

5
mobile/lib/shared/read_state/read_state_provider.dart
View File

@@ -75,7 +75,10 @@ class ReadStateNotifier extends Notifier<ReadStateState> {
_forcedUnreadContexts.clear();
final relayConfig = ref.watch(relayConfigProvider);
ref.watch(relaySessionProvider);
// Watch the session notifier, not its state: a status change (app resume,
// reconnect) must not rebuild the manager and drop session-local forced
// unread flags. The listener below re-syncs on reconnect instead.
ref.watch(relaySessionProvider.notifier);
ref.watch(activeCommunityProvider);
final nsec = relayConfig.nsec?.trim();

153
mobile/test/features/channels/read_state/read_state_manager_test.dart
View File

@@ -557,6 +557,134 @@ void main() {
expect(manager.getEffectiveTimestamp('channel-1'), 300);
});
test(
'a connect during a successful initialize does not sync again',
() async {
SharedPreferences.setMockInitialValues({});
final prefs = await SharedPreferences.getInstance();
final keychain = nostr.Keys.generate();
final crypto = ReadStateCrypto.tryCreate(
nsec: keychain.nsec,
pubkey: keychain.public,
)!;
// First connect: initialize waits on a history REQ held for the socket
// while the session reports connected and the provider's listener fires.
final gate = Completer<void>();
final relay = _FakeRelaySession()..fetchGate = gate.future;
final manager = ReadStateManager(
pubkey: keychain.public,
relayUrl: _relayUrl,
prefs: prefs,
crypto: crypto,
relaySession: relay,
signedEventRelay: _FakeSignedEventRelay(),
remoteEnabled: true,
onChanged: () {},
);
final initializing = manager.initialize();
final reinitializing = manager.reinitializeRemote();
relay.fetchGate = null;
gate.complete();
await Future.wait([initializing, reinitializing]);
expect(relay.fetchCount, 1);
expect(relay.subscribeCount, 1);
expect(relay.liveSubscriptions, 1);
// A later reconnect syncs again and still leaves one subscription.
await manager.reinitializeRemote();
expect(relay.fetchCount, 2);
expect(relay.subscribeCount, 2);
expect(relay.liveSubscriptions, 1);
manager.dispose();
expect(relay.liveSubscriptions, 0);
},
);
test('a connect during a failed initialize syncs again', () async {
SharedPreferences.setMockInitialValues({});
final prefs = await SharedPreferences.getInstance();
final keychain = nostr.Keys.generate();
final crypto = ReadStateCrypto.tryCreate(
nsec: keychain.nsec,
pubkey: keychain.public,
)!;
final gate = Completer<void>();
final relay = _FakeRelaySession()
..fetchGate = gate.future
..failNextFetch = true;
final manager = ReadStateManager(
pubkey: keychain.public,
relayUrl: _relayUrl,
prefs: prefs,
crypto: crypto,
relaySession: relay,
signedEventRelay: _FakeSignedEventRelay(),
remoteEnabled: true,
onChanged: () {},
);
final initializing = manager.initialize();
final reinitializing = manager.reinitializeRemote();
relay.fetchGate = null;
gate.complete();
await Future.wait([initializing, reinitializing]);
// Both passes subscribed, but the later one replaced the earlier one.
expect(relay.fetchCount, 2);
expect(relay.subscribeCount, 2);
expect(relay.liveSubscriptions, 1);
manager.dispose();
expect(relay.liveSubscriptions, 0);
});
test('reconnects during a re-sync leave one live subscription', () async {
SharedPreferences.setMockInitialValues({});
final prefs = await SharedPreferences.getInstance();
final keychain = nostr.Keys.generate();
final crypto = ReadStateCrypto.tryCreate(
nsec: keychain.nsec,
pubkey: keychain.public,
)!;
final relay = _FakeRelaySession();
final manager = ReadStateManager(
pubkey: keychain.public,
relayUrl: _relayUrl,
prefs: prefs,
crypto: crypto,
relaySession: relay,
signedEventRelay: _FakeSignedEventRelay(),
remoteEnabled: true,
onChanged: () {},
);
await manager.initialize();
// The first re-sync holds its fetch; three more connects arrive.
final gate = Completer<void>();
relay.fetchGate = gate.future;
final first = manager.reinitializeRemote();
await Future<void>.delayed(Duration.zero);
final later = [
manager.reinitializeRemote(),
manager.reinitializeRemote(),
manager.reinitializeRemote(),
];
relay.fetchGate = null;
gate.complete();
await Future.wait([first, ...later]);
// One initial pass, the held re-sync, and one shared re-sync after it.
expect(relay.fetchCount, 3);
expect(relay.subscribeCount, 3);
expect(relay.liveSubscriptions, 1);
manager.dispose();
expect(relay.liveSubscriptions, 0);
});
test('remote read-state rollback is ignored', () async {
SharedPreferences.setMockInitialValues({});
final prefs = await SharedPreferences.getInstance();
@@ -741,17 +869,38 @@ NostrEvent _readStateEvent({
class _FakeRelaySession extends RelaySessionNotifier {
List<NostrEvent> historyEvents = [];
Future<void>? fetchGate;
bool failNextFetch = false;
int fetchCount = 0;
int subscribeCount = 0;
int liveSubscriptions = 0;
@override
Future<List<NostrEvent>> fetchHistory(
NostrFilter filter, {
Duration timeout = const Duration(seconds: 8),
}) async => historyEvents;
}) async {
fetchCount++;
await fetchGate;
if (failNextFetch) {
failNextFetch = false;
throw TimeoutException('history timed out');
}
return historyEvents;
}
@override
Future<void Function()> subscribe(
NostrFilter filter,
void Function(NostrEvent) onEvent, {
void Function(String message)? onClosed,
}) async => () {};
}) async {
subscribeCount++;
liveSubscriptions++;
var open = true;
return () {
if (open) liveSubscriptions--;
open = false;
};
}
}

123
mobile/test/features/channels/read_state/read_state_provider_test.dart
View File

@@ -24,6 +24,7 @@ void main() {
late ProviderContainer container;
late _FakeRelayConfig config;
late _FakeRelaySession session;
Future<ReadStateNotifier> pumpNotifier({
Map<String, Object> Function(String pubkey)? stored,
@@ -53,11 +54,19 @@ void main() {
await Future<void>.delayed(Duration.zero);
}
expect(container.read(readStateProvider).isReady, isTrue);
session =
container.read(relaySessionProvider.notifier) as _FakeRelaySession;
return notifier;
}
ReadStateState state() => container.read(readStateProvider);
Future<void> flushMicrotasks() async {
for (var i = 0; i < 10; i++) {
await Future<void>.delayed(Duration.zero);
}
}
Future<void> settle() async {
for (var i = 0; i < 10 && !state().isReady; i++) {
await Future<void>.delayed(Duration.zero);
@@ -98,6 +107,86 @@ void main() {
expect(state().effectiveTimestamp(channelId), isNull);
});
test('a disconnect and reconnect keeps the manager and forced unread '
'flags, and resyncs once on connect', () async {
final notifier = await pumpNotifier();
// Persists on the same manager, so it survives only if that manager does.
notifier.markContextRead(channelId, 500);
notifier.markContextUnread(msgKey, channelId: channelId);
expect(session.fetchCount, 1);
expect(session.subscribeCount, 1);
final version = state().version;
for (final status in [
SessionStatus.disconnected,
SessionStatus.connecting,
SessionStatus.reconnecting,
]) {
session.setStatus(status);
await flushMicrotasks();
// No rebuild: still ready, flags kept, and nothing refetched.
expect(state().isReady, isTrue, reason: status.name);
expect(state().isForcedUnread(msgKey), isTrue, reason: status.name);
expect(state().effectiveTimestamp(channelId), 500);
expect(session.fetchCount, 1, reason: status.name);
expect(session.subscribeCount, 1, reason: status.name);
}
session.setStatus(SessionStatus.connected);
await flushMicrotasks();
// The reconnect listener refetches and replaces the live subscription
// on the same manager.
expect(session.fetchCount, 2);
expect(session.subscribeCount, 2);
expect(session.liveSubscriptions, 1);
expect(state().isReady, isTrue);
expect(state().isForcedUnread(msgKey), isTrue);
expect(state().locallyForcedChannelIds, {channelId});
expect(state().effectiveTimestamp(channelId), 500);
expect(state().version, greaterThanOrEqualTo(version));
// A second reconnect resyncs the same way.
session.setStatus(SessionStatus.disconnected);
session.setStatus(SessionStatus.connected);
await flushMicrotasks();
expect(session.fetchCount, 3);
expect(session.liveSubscriptions, 1);
expect(state().isForcedUnread(msgKey), isTrue);
});
test('switching account rebuilds the manager and drops forced unread '
'flags', () async {
final notifier = await pumpNotifier();
notifier.markContextUnread(msgKey, channelId: channelId);
expect(session.fetchCount, 1);
final pubkey = state().pubkey;
config.switchAccount(nostr.Keys.generate().nsec);
container.read(readStateProvider);
await settle();
expect(state().pubkey, isNot(pubkey));
expect(state().forcedUnreadContexts, isEmpty);
expect(session.fetchCount, 2);
expect(session.subscribeCount, 2);
expect(session.liveSubscriptions, 1);
});
test('switching community rebuilds the manager and drops forced unread '
'flags', () async {
final notifier = await pumpNotifier();
notifier.markContextUnread(msgKey, channelId: channelId);
config.switchTo('http://localhost:2');
container.read(readStateProvider);
await settle();
expect(state().forcedUnreadContexts, isEmpty);
expect(session.fetchCount, 2);
expect(session.liveSubscriptions, 1);
});
test('message unread → read → unread round-trips through the real '
'notifier', () async {
final notifier = await pumpNotifier();
@@ -189,7 +278,7 @@ void main() {
}
class _FakeRelayConfig extends RelayConfigNotifier {
final String nsec;
String nsec;
_FakeRelayConfig(this.nsec);
@@ -198,11 +287,23 @@ class _FakeRelayConfig extends RelayConfigNotifier {
void switchTo(String baseUrl) =>
state = RelayConfig(baseUrl: baseUrl, nsec: nsec);
void switchAccount(String newNsec) {
nsec = newNsec;
state = RelayConfig(baseUrl: state.baseUrl, nsec: newNsec);
}
}
/// Relay session that never connects and returns no history, keeping the
/// manager fully local while exercising its real code paths.
/// Relay session that returns no history and counts the read-state fetches
/// and live subscriptions the manager makes, keeping it fully local while
/// exercising its real code paths. Its status is driven by [setStatus].
class _FakeRelaySession extends RelaySessionNotifier {
int fetchCount = 0;
int subscribeCount = 0;
int liveSubscriptions = 0;
void setStatus(SessionStatus status) => state = SessionState(status: status);
@override
SessionState build() =>
const SessionState(status: SessionStatus.disconnected);
@@ -211,14 +312,26 @@ class _FakeRelaySession extends RelaySessionNotifier {
Future<List<NostrEvent>> fetchHistory(
NostrFilter filter, {
Duration timeout = const Duration(seconds: 8),
}) async => [];
}) async {
// The manager's pre-publish own-slot lookup is not a full history fetch.
if (filter.tags.containsKey('#t')) fetchCount++;
return [];
}
@override
Future<void Function()> subscribe(
NostrFilter filter,
void Function(NostrEvent) onEvent, {
void Function(String message)? onClosed,
}) async => () {};
}) async {
subscribeCount++;
liveSubscriptions++;
var open = true;
return () {
if (open) liveSubscriptions--;
open = false;
};
}
}
class _FakeAppLifecycle extends AppLifecycleNotifier {