comunes/tane
1
0
Fork 0
Code Issues Pull requests Projects Releases 7 Packages Wiki Activity Actions

fix(market): recover by itself when the shared connection comes up

Browse source 
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)
This commit is contained in:
vjrj 2026-07-22 12:57:24 +02:00
parent 981cf10048
commit 8a7dc8d3dc
6 changed files with 273 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

51
apps/app_seeds/lib/services/social_connection.dart
View file

@ -25,19 +25,36 @@ class SocialConnection {
required SocialSettings settings,
SessionOpener? open,
Stream<bool>? online,
List<Duration>? 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<bool>? _online;
final List<Duration> _retrySchedule;
final _sessions = StreamController<SocialSession?>.broadcast();
SocialSession? _current;
Future<SocialSession?>? _pending;
StreamSubscription<bool>? _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<SocialSession?> 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<void> dispose() async {
_disposed = true;
_cancelRetry();
await _onlineSub?.cancel();
_onlineSub = null;
_drop();

42
apps/app_seeds/lib/state/offers_cubit.dart
View file

@ -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<OffersState> {
OffersCubit(
this._transport, {
SocialConnection? connection,
Future<Uint8List?> Function(String varietyId)? coverPhoto,
String? Function(Uint8List bytes)? thumbnail,
Future<void> 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<SocialSession?>? _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<OffersState> {
/// from now on — older results arrive via [loadNextPage], not by draining the
/// whole area into memory.
Future<void> 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<OffersState> {
@override
Future<void> close() async {
_searchTimeout?.cancel();
await _connSub?.cancel();
await _sub?.cancel();
await _onDispose?.call();
return super.close();
@ -484,6 +521,7 @@ Future<OffersCubit> createOffersCubit(
final session = await connection.session();
return OffersCubit(
session?.offers,
connection: connection, // keeps following (re)connects — see _onSession
coverPhoto: repository?.coverPhotoFor,
thumbnail: offerThumbnailDataUri,
);

6
apps/app_seeds/lib/ui/market_screen.dart
View file

@ -122,10 +122,10 @@ class _MarketScreenState extends State<MarketScreen> {
_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<MarketScreen> {
}
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));
}
}

65
apps/app_seeds/test/services/social_connection_test.dart
View file

@ -45,11 +45,13 @@ void main() {
required List<FakeChannel> opened,
Stream<bool>? online,
bool Function()? fail,
List<Duration>? 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 = <FakeChannel>[];
var down = true;
final conn = make(
opened: opened,
online: const Stream.empty(), // connectivity never speaks
fail: () => down,
retrySchedule: const [Duration(milliseconds: 5)],
);
final emitted = <SocialSession?>[];
conn.sessions.listen(emitted.add);
conn.start();
await Future<void>.delayed(Duration.zero);
expect(conn.current, isNull); // first attempt failed
down = false; // network path recovers
await Future<void>.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 = <FakeChannel>[];
final conn = make(
opened: opened,
fail: () => true,
retrySchedule: const [Duration(milliseconds: 5)],
);
expect(await conn.session(), isNull);
await Future<void>.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 = <FakeChannel>[];
var down = true;
final conn = make(
opened: opened,
online: const Stream.empty(),
fail: () => down,
retrySchedule: const [Duration(milliseconds: 20)],
);
conn.start();
await Future<void>.delayed(Duration.zero); // first attempt fails
await conn.dispose(); // cancels the scheduled retry
down = false;
await Future<void>.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 = <FakeChannel>[];

111
apps/app_seeds/test/state/offers_cubit_test.dart
View file

@ -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<bool>? 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<void>.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<void>.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<bool>.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<Event> subscribe(Filter filter) => const Stream.empty();
@override
Future<List<Event>> reqOnce(Filter filter) async => const [];
@override
Future<void> close() async {}
}

5
packages/commons_core/lib/src/social/nostr/nostr_offer_transport.dart
View file

@ -46,9 +46,10 @@ class NostrOfferTransport implements OfferTransport {
@override
Future<OfferPage> 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.
// 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 = <Offer>[];
for (final e in events) {