fix(mobile): recover discovery subscriptions and coordinate deletions

Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-09-09 12:59:55 -04:00
parent f9d2961264
commit 1eabb049c6
3 changed files with 147 additions and 7 deletions

View File

@@ -51,18 +51,23 @@ Future<Set<String>> _ownedAgentKeys(
// Separate subscription owner: a refresh must not tear down/replay its trigger.
class _AgentDirectoryUpdates extends Notifier<int> {
Object? failure;
@override
int build() {
final sessionState = ref.watch(relaySessionProvider);
ref.watch(myPubkeyProvider);
ref.watch(relayConfigProvider);
failure = null;
var disposed = false;
var attempts = 0;
Timer? retry;
void Function()? unsubscribe;
Timer? debounce;
ref.onDispose(() {
disposed = true;
unsubscribe?.call();
debounce?.cancel();
retry?.cancel();
});
void changed() {
if (disposed || debounce != null) return;
@@ -74,17 +79,26 @@ class _AgentDirectoryUpdates extends Notifier<int> {
if (sessionState.status == SessionStatus.connected) {
final session = ref.read(relaySessionProvider.notifier);
unawaited(() async {
Future<void> subscribe() async {
if (disposed) return;
attempts++;
try {
final close = await session.subscribeWithStatus(
NostrFilter(
kinds: const [0, 10100, 30177, 39002],
kinds: const [0, 5, 10100, 30177, 39002],
limit: 0,
since: DateTime.now().millisecondsSinceEpoch ~/ 1000,
),
(_) => changed(),
(event) {
if (event.kind == 5 && !_isAgentCoordinateDeletion(event)) return;
changed();
},
onClosed: (_) => changed(),
onStatusChanged: (_) => changed(),
onStatusChanged: (status) {
if (disposed) return;
if (status == RelaySubscriptionStatus.ready) failure = null;
changed();
},
);
if (disposed) {
close();
@@ -92,10 +106,22 @@ class _AgentDirectoryUpdates extends Notifier<int> {
unsubscribe = close;
}
} catch (error) {
// Suggestions may age during failure; publication always reads fresh.
debugPrint('[AgentDirectory] refresh subscription failed: $error');
if (disposed) return;
failure = error;
changed();
// Establishment can fail before the session installs status callbacks.
// Three attempts per session generation; exhausted failure stays
// visible to directory readers until reconnect/rebuild permits retry.
if (attempts < 3) {
retry = Timer(Duration(milliseconds: 250 * attempts), () {
retry = null;
unawaited(subscribe());
});
}
}
}());
}
unawaited(subscribe());
}
return 0;
}
@@ -103,3 +129,15 @@ class _AgentDirectoryUpdates extends Notifier<int> {
final _agentDirectoryUpdatesProvider =
NotifierProvider<_AgentDirectoryUpdates, int>(_AgentDirectoryUpdates.new);
// Desktop build_agent_delete emits a kind:5 a-coordinate, not a 30177 event.
bool _isAgentCoordinateDeletion(NostrEvent event) =>
verifySignedEvent(event) &&
event.tags.any((tag) {
if (tag.length < 2 || tag[0] != 'a') return false;
final coordinate = tag[1].split(':');
return coordinate.length == 3 &&
coordinate[0] == '30177' &&
coordinate[1] == event.pubkey &&
RegExp(r'^[0-9a-f]{64}$').hasMatch(coordinate[2]);
});

View File

@@ -74,6 +74,8 @@ final agentDirectoryProvider = FutureProvider<List<AgentDirectoryEntry>>((
ref,
) async {
ref.watch(_agentDirectoryUpdatesProvider);
final failure = ref.read(_agentDirectoryUpdatesProvider.notifier).failure;
if (failure != null) throw failure;
final sessionState = ref.watch(relaySessionProvider);
if (sessionState.status != SessionStatus.connected) return const [];
final session = ref.read(relaySessionProvider.notifier);

View File

@@ -13,6 +13,13 @@ class _Session extends PolicySession {
void Function(NostrEvent)? changed;
void Function(RelaySubscriptionStatus)? status;
int closed = 0;
int attempts = 0;
int failures = 0;
NostrFilter? subscription;
void deliver(NostrEvent event) {
if (subscription?.kinds.contains(event.kind) == true) changed?.call(event);
}
List<NostrEvent> Function(NostrFilter)? ownedPage;
@override
Future<String> fetchRelaySelf() async => authority;
@@ -23,8 +30,12 @@ class _Session extends PolicySession {
void Function(String)? onClosed,
required void Function(RelaySubscriptionStatus) onStatusChanged,
}) async {
attempts++;
if (attempts <= failures) throw StateError('establishment unavailable');
subscription = filter;
changed = onEvent;
status = onStatusChanged;
onStatusChanged(RelaySubscriptionStatus.ready);
return () {
closed++;
};
@@ -151,6 +162,95 @@ void main() {
},
);
test(
'failed establishment recovers while connected; coordinate delete refreshes',
() async {
final events = [owned, policy];
final session = _Session(events, relay.public)..failures = 1;
final c = container(session);
addTearDown(c.dispose);
await c.read(agentDirectoryProvider.future);
await Future<void>.delayed(const Duration(milliseconds: 180));
await expectLater(
c.read(agentDirectoryProvider.future),
throwsStateError,
);
await Future<void>.delayed(const Duration(milliseconds: 300));
expect(session.attempts, 2);
expect((await c.read(agentDirectoryProvider.future)).length, 1);
final deny = signed(
owner,
30177,
{'name': 'Remote helper', 'parallelism': 1, 'respond_to': 'nobody'},
time: 101,
tags: [
['d', agent.public],
],
);
events.add(deny);
session.deliver(deny);
await Future<void>.delayed(const Duration(milliseconds: 180));
expect(
(await c.read(agentDirectoryProvider.future)).single.respondTo,
'nobody',
);
final deletion = signed(
owner,
5,
'',
tags: [
['a', '30177:${owner.public}:${agent.public}'],
],
);
events.remove(policy);
events.remove(deny);
session.deliver(
signed(
owner,
5,
'',
tags: [
['a', '30000:${owner.public}:other'],
],
),
);
await Future<void>.delayed(const Duration(milliseconds: 180));
expect(
(await c.read(agentDirectoryProvider.future)).single.respondTo,
'nobody',
);
session.deliver(deletion);
await Future<void>.delayed(const Duration(milliseconds: 180));
expect(await c.read(agentDirectoryProvider.future), isEmpty);
session.status!(RelaySubscriptionStatus.ready);
session.deliver(deletion); // replay remains removed after reconnect
await Future<void>.delayed(const Duration(milliseconds: 180));
expect(await c.read(agentDirectoryProvider.future), isEmpty);
},
);
test(
'establishment retries are bounded and retired scopes cannot retry',
() async {
final session = _Session([owned, policy], relay.public)..failures = 99;
final c = container(session);
await c.read(agentDirectoryProvider.future);
await Future<void>.delayed(const Duration(milliseconds: 1100));
expect(session.attempts, 3);
await expectLater(
c.read(agentDirectoryProvider.future),
throwsStateError,
);
c.dispose();
final retired = _Session([owned, policy], relay.public)..failures = 99;
final c2 = container(retired);
await c2.read(agentDirectoryProvider.future);
c2.dispose();
await Future<void>.delayed(const Duration(milliseconds: 300));
expect(retired.attempts, 1);
},
);
test(
'owned coordinates paginate equal-time pages and reject a stalled cursor',
() async {