From 8a7dc8d3dc066135641a7840fd3349c9a2ccd9e1 Mon Sep 17 00:00:00 2001 From: vjrj Date: Wed, 22 Jul 2026 12:57:24 +0200 Subject: [PATCH] fix(market): recover by itself when the shared connection comes up Fresh installs could sit on 'can't reach the servers' forever: the offers cubit captured the transport once at build time, the shared connection only retried on a connectivity CHANGE, and a silently-filtered relay could stall the pool for minutes. - SocialConnection: retry with backoff after a failed attempt while started and not knowingly offline (injectable schedule for tests) - OffersCubit: follow connection.sessions, re-attach the transport and re-run the last discovery on (re)connect; announce drops via connectionEpoch - market _init: record the wanted area on the cubit even while offline - NostrOfferTransport.discoverPage: sort a copy (channel lists may be unmodifiable) --- .../lib/services/social_connection.dart | 51 +++++++- apps/app_seeds/lib/state/offers_cubit.dart | 42 ++++++- apps/app_seeds/lib/ui/market_screen.dart | 6 +- .../test/services/social_connection_test.dart | 65 ++++++++++ .../test/state/offers_cubit_test.dart | 111 ++++++++++++++++++ .../social/nostr/nostr_offer_transport.dart | 7 +- 6 files changed, 273 insertions(+), 9 deletions(-) diff --git a/apps/app_seeds/lib/services/social_connection.dart b/apps/app_seeds/lib/services/social_connection.dart index 8e57949..332621c 100644 --- a/apps/app_seeds/lib/services/social_connection.dart +++ b/apps/app_seeds/lib/services/social_connection.dart @@ -25,19 +25,36 @@ class SocialConnection { required SocialSettings settings, SessionOpener? open, Stream? online, + List? retrySchedule, }) : _settings = settings, _open = open ?? social.openSession, - _online = online; + _online = online, + _retrySchedule = retrySchedule ?? _defaultRetrySchedule; + + /// Backoff for self-retries after a failed attempt while started and not + /// knowingly offline (the fresh-install case: online the whole time, first + /// connect fails, so no connectivity change ever retriggers a connect). + static const _defaultRetrySchedule = [ + Duration(seconds: 5), + Duration(seconds: 15), + Duration(seconds: 45), + Duration(seconds: 90), + ]; final SocialSettings _settings; final SessionOpener _open; final Stream? _online; + final List _retrySchedule; final _sessions = StreamController.broadcast(); SocialSession? _current; Future? _pending; StreamSubscription? _onlineSub; bool _disposed = false; + bool _started = false; + bool _knownOffline = false; + Timer? _retryTimer; + int _retryIndex = 0; /// Emits the live session on each (re)connect, and null when it drops. Stream get sessions => _sessions.stream; @@ -48,10 +65,14 @@ class SocialConnection { /// Begins watching connectivity (reconnect on regain, drop when offline) and /// attempts an initial connect. Idempotent-ish; call once at startup. void start() { + _started = true; _onlineSub = (_online ?? _connectivityOnline()).listen((isOnline) { + _knownOffline = !isOnline; if (!isOnline) { + _cancelRetry(); _drop(); } else if (_current == null) { + _retryIndex = 0; // fresh network — start the backoff over unawaited(session()); } }); @@ -75,15 +96,40 @@ class SocialConnection { return null; } _current = s; + _retryIndex = 0; + _cancelRetry(); _sessions.add(s); return s; } catch (_) { - return null; // unreachable — a later connectivity change retries + _scheduleRetry(); // unreachable — retry with backoff (see below) + return null; } finally { _pending = null; } } + /// After a failed attempt, retries by itself while started and not knowingly + /// offline — a connectivity change may never come if the device was online + /// all along. Un-started connections (one-shot [session] callers, tests) + /// never leave a timer behind. + void _scheduleRetry() { + if (!_started || _disposed || _knownOffline || _current != null) return; + _cancelRetry(); + final i = _retryIndex < _retrySchedule.length + ? _retryIndex + : _retrySchedule.length - 1; + _retryIndex++; + _retryTimer = Timer(_retrySchedule[i], () { + if (_disposed || _current != null) return; + unawaited(session()); + }); + } + + void _cancelRetry() { + _retryTimer?.cancel(); + _retryTimer = null; + } + void _drop() { final s = _current; _current = null; @@ -95,6 +141,7 @@ class SocialConnection { Future dispose() async { _disposed = true; + _cancelRetry(); await _onlineSub?.cancel(); _onlineSub = null; _drop(); diff --git a/apps/app_seeds/lib/state/offers_cubit.dart b/apps/app_seeds/lib/state/offers_cubit.dart index d6c462b..60e1c36 100644 --- a/apps/app_seeds/lib/state/offers_cubit.dart +++ b/apps/app_seeds/lib/state/offers_cubit.dart @@ -10,6 +10,7 @@ import '../services/offer_mapper.dart'; import '../services/offer_outbox.dart'; import '../services/offer_thumbnail.dart'; import '../services/social_connection.dart'; +import '../services/social_service.dart'; /// State of the offer discovery/publish screen. Transport-agnostic — it holds /// only what the UI shows, never a relay handle. @@ -28,6 +29,7 @@ class OffersState extends Equatable { this.nextPageCursor, this.publishing = false, this.hasSearched = false, + this.connectionEpoch = 0, this.error, }); @@ -80,6 +82,11 @@ class OffersState extends Equatable { /// from "searched, found nothing". final bool hasSearched; + /// Bumped whenever the underlying transport comes or goes, so the UI rebuilds + /// and re-reads [OffersCubit.isOnline] — states differing only here are + /// otherwise Equatable-equal and bloc would skip the emit. + final int connectionEpoch; + /// Last error, in human terms for the UI (null when fine). final String? error; @@ -136,6 +143,7 @@ class OffersState extends Equatable { int? Function()? nextPageCursor, bool? publishing, bool? hasSearched, + int? connectionEpoch, String? Function()? error, }) { return OffersState( @@ -153,6 +161,7 @@ class OffersState extends Equatable { nextPageCursor != null ? nextPageCursor() : this.nextPageCursor, publishing: publishing ?? this.publishing, hasSearched: hasSearched ?? this.hasSearched, + connectionEpoch: connectionEpoch ?? this.connectionEpoch, error: error != null ? error() : this.error, ); } @@ -172,6 +181,7 @@ class OffersState extends Equatable { nextPageCursor, publishing, hasSearched, + connectionEpoch, error, ]; } @@ -183,15 +193,40 @@ class OffersState extends Equatable { class OffersCubit extends Cubit { OffersCubit( this._transport, { + SocialConnection? connection, Future Function(String varietyId)? coverPhoto, String? Function(Uint8List bytes)? thumbnail, Future Function()? onDispose, }) : _coverPhoto = coverPhoto, _thumbnail = thumbnail, _onDispose = onDispose, - super(const OffersState()); + super(const OffersState()) { + // Fresh-install fix: the transport captured at build time may be null while + // the shared connection is still coming up (or dropped). Follow the + // connection so the market recovers by itself instead of staying offline + // until a manual retry. + _connSub = connection?.sessions.listen(_onSession); + } - final OfferTransport? _transport; + OfferTransport? _transport; + StreamSubscription? _connSub; + + /// The last area prefix asked of [discover]; re-run when the connection + /// (re)appears so results show without the user tapping anything. + String? _lastPrefix; + + void _onSession(SocialSession? session) { + if (isClosed) return; + final transport = session?.offers; + if (identical(transport, _transport)) return; + _transport = transport; + final prefix = _lastPrefix; + if (transport != null && prefix != null) { + unawaited(discover(prefix)); // emits fresh states as results arrive + } else { + emit(state.copyWith(connectionEpoch: state.connectionEpoch + 1)); + } + } /// Fetches a variety's cover photo bytes; null in tests or when no inventory /// repo is wired. @@ -225,6 +260,7 @@ class OffersCubit extends Cubit { /// from now on — older results arrive via [loadNextPage], not by draining the /// whole area into memory. Future discover(String geohashPrefix) async { + _lastPrefix = geohashPrefix; final transport = _transport; if (transport == null) { emit(state.copyWith(error: () => 'offline', hasSearched: true)); @@ -467,6 +503,7 @@ class OffersCubit extends Cubit { @override Future close() async { _searchTimeout?.cancel(); + await _connSub?.cancel(); await _sub?.cancel(); await _onDispose?.call(); return super.close(); @@ -484,6 +521,7 @@ Future createOffersCubit( final session = await connection.session(); return OffersCubit( session?.offers, + connection: connection, // keeps following (re)connects — see _onSession coverPhoto: repository?.coverPhotoFor, thumbnail: offerThumbnailDataUri, ); diff --git a/apps/app_seeds/lib/ui/market_screen.dart b/apps/app_seeds/lib/ui/market_screen.dart index 23941a5..5b6e0ed 100644 --- a/apps/app_seeds/lib/ui/market_screen.dart +++ b/apps/app_seeds/lib/ui/market_screen.dart @@ -122,10 +122,10 @@ class _MarketScreenState extends State { _loading = false; }); await previous?.close(); - if (area.isNotEmpty && cubit.isOnline) { + if (area.isNotEmpty) { // Flush anything parked while offline, then show what's out there. final outbox = widget.outbox; - if (outbox != null && repo != null) { + if (outbox != null && repo != null && cubit.isOnline) { await flushOutbox( outbox: outbox, cubit: cubit, @@ -136,6 +136,8 @@ class _MarketScreenState extends State { } if (mounted) { final precision = await widget.settings.searchPrecision(); + // Even when offline: this records the wanted area on the cubit, which + // re-runs the discovery by itself once the connection comes up. if (mounted) await cubit.discover(searchPrefix(area, precision)); } } diff --git a/apps/app_seeds/test/services/social_connection_test.dart b/apps/app_seeds/test/services/social_connection_test.dart index e4c7151..7b856de 100644 --- a/apps/app_seeds/test/services/social_connection_test.dart +++ b/apps/app_seeds/test/services/social_connection_test.dart @@ -45,11 +45,13 @@ void main() { required List opened, Stream? online, bool Function()? fail, + List? retrySchedule, }) => SocialConnection( social: social, settings: settings, online: online, + retrySchedule: retrySchedule, open: (_) async { if (fail?.call() ?? false) throw StateError('unreachable'); final ch = FakeChannel(); @@ -106,6 +108,69 @@ void main() { await online.close(); }); + test( + 'started connection retries by itself after a failed first attempt, ' + 'without any connectivity event', () async { + // The fresh-install case: device online the whole time, but the very first + // connect fails (cold DNS/TLS). The connectivity stream never fires, so + // only the backoff retry can bring the market up without a manual Retry. + final opened = []; + var down = true; + final conn = make( + opened: opened, + online: const Stream.empty(), // connectivity never speaks + fail: () => down, + retrySchedule: const [Duration(milliseconds: 5)], + ); + final emitted = []; + conn.sessions.listen(emitted.add); + + conn.start(); + await Future.delayed(Duration.zero); + expect(conn.current, isNull); // first attempt failed + + down = false; // network path recovers + await Future.delayed(const Duration(milliseconds: 100)); + expect(conn.current, isNotNull, reason: 'backoff retry reconnected'); + expect(opened, hasLength(1)); + expect(emitted.last, isNotNull, reason: 'recovery announced on sessions'); + + await conn.dispose(); + }); + + test('a plain session() failure schedules no retry when never started', + () async { + // Widget tests build cubits against an un-started connection; a failed + // one-shot session() must not leave a pending retry timer behind. + final opened = []; + final conn = make( + opened: opened, + fail: () => true, + retrySchedule: const [Duration(milliseconds: 5)], + ); + expect(await conn.session(), isNull); + await Future.delayed(const Duration(milliseconds: 50)); + expect(opened, isEmpty); // no background retry fired + await conn.dispose(); + }); + + test('dispose cancels a pending backoff retry', () async { + final opened = []; + var down = true; + final conn = make( + opened: opened, + online: const Stream.empty(), + fail: () => down, + retrySchedule: const [Duration(milliseconds: 20)], + ); + conn.start(); + await Future.delayed(Duration.zero); // first attempt fails + await conn.dispose(); // cancels the scheduled retry + down = false; + await Future.delayed(const Duration(milliseconds: 100)); + expect(opened, isEmpty, reason: 'no reconnect after dispose'); + }); + test('returns null when the relay is unreachable, and retries later', () async { final opened = []; diff --git a/apps/app_seeds/test/state/offers_cubit_test.dart b/apps/app_seeds/test/state/offers_cubit_test.dart index 650088a..4c84dc6 100644 --- a/apps/app_seeds/test/state/offers_cubit_test.dart +++ b/apps/app_seeds/test/state/offers_cubit_test.dart @@ -7,7 +7,11 @@ import 'package:tane/data/variety_repository.dart'; import 'package:tane/db/enums.dart'; import 'package:tane/services/discovery_area.dart'; import 'package:tane/services/offer_mapper.dart'; +import 'package:nostr/nostr.dart'; import 'package:tane/services/offer_outbox.dart'; +import 'package:tane/services/social_connection.dart'; +import 'package:tane/services/social_service.dart'; +import 'package:tane/services/social_settings.dart'; import 'package:tane/state/offers_cubit.dart'; import '../support/test_support.dart'; @@ -835,4 +839,111 @@ void main() { await cubit.close(); }); }); + + group('connection recovery (fresh-install fix)', () { + const seedHex = + '000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f'; + + late SocialService social; + late SocialSettings settings; + + setUp(() async { + social = await SocialService.fromRootSeedHex(seedHex); + settings = SocialSettings(InMemorySecretStore()); + }); + + SocialConnection makeConnection({ + required bool Function() down, + Stream? online, + }) => + SocialConnection( + social: social, + settings: settings, + online: online ?? const Stream.empty(), + retrySchedule: const [Duration(milliseconds: 5)], + open: (_) async { + if (down()) throw StateError('unreachable'); + return SocialSession(_NoopChannel()); + }, + ); + + test('comes online by itself when the shared connection connects later', + () async { + var down = true; + final conn = makeConnection(down: () => down); + final cubit = await createOffersCubit(conn); + expect(cubit.isOnline, isFalse); // built while unreachable + + final epochBefore = cubit.state.connectionEpoch; + conn.start(); + down = false; // network path recovers; backoff retry reconnects + await Future.delayed(const Duration(milliseconds: 100)); + + expect(cubit.isOnline, isTrue, reason: 'no manual Retry needed'); + expect(cubit.state.connectionEpoch, greaterThan(epochBefore), + reason: 'a state was emitted so the UI re-reads isOnline'); + + await cubit.close(); + await conn.dispose(); + }); + + test('re-runs the last discovery when the connection recovers', () async { + var down = true; + final conn = makeConnection(down: () => down); + final cubit = await createOffersCubit(conn); + + await cubit.discover('sp3e9'); // offline → error, but the wish is kept + expect(cubit.state.error, 'offline'); + + conn.start(); + down = false; + await Future.delayed(const Duration(milliseconds: 100)); + + expect(cubit.isOnline, isTrue); + expect(cubit.state.areaGeohash, 'sp3e9', reason: 'discovery re-ran'); + expect(cubit.state.hasSearched, isTrue); + expect(cubit.state.error, isNull, reason: 'offline error cleared'); + + await cubit.close(); + await conn.dispose(); + }); + + test('goes offline (and tells the UI) when the session drops', () async { + final online = StreamController.broadcast(); + final conn = makeConnection(down: () => false, online: online.stream); + conn.start(); + await pumpEventQueue(); + final cubit = await createOffersCubit(conn); + expect(cubit.isOnline, isTrue); + + final epochBefore = cubit.state.connectionEpoch; + online.add(false); // network lost + await pumpEventQueue(); + + expect(cubit.isOnline, isFalse); + expect(cubit.state.connectionEpoch, greaterThan(epochBefore)); + + await cubit.close(); + await conn.dispose(); + await online.close(); + }); + }); +} + +/// A no-op [NostrChannel]: just enough to build a [SocialSession] whose offer +/// transport answers with empty results. +class _NoopChannel implements NostrChannel { + @override + String get privateKeyHex => '00' * 32; + @override + String get publicKeyHex => 'ab' * 32; + @override + Future<({bool accepted, String message})> publish(Event event) async => + (accepted: true, message: ''); + @override + Stream subscribe(Filter filter) => const Stream.empty(); + @override + Future> reqOnce(Filter filter) async => const []; + @override + Future close() async {} } diff --git a/packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart b/packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart index 305f842..bd4c9d8 100644 --- a/packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart +++ b/packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart @@ -46,9 +46,10 @@ class NostrOfferTransport implements OfferTransport { @override Future discoverPage(DiscoveryQuery query) async { - final events = await _conn.reqOnce(_filter(query)) - // Newest first; the relays dedup within themselves and reqOnce dedups - // across them, but order is not guaranteed, so sort here. + // Newest first; the relays dedup within themselves and reqOnce dedups + // across them, but order is not guaranteed, so sort here — on a copy, as + // the channel may hand out an unmodifiable list. + final events = [...await _conn.reqOnce(_filter(query))] ..sort((a, b) => b.createdAt.compareTo(a.createdAt)); final offers = []; for (final e in events) {