diff --git a/mobile/lib/shared/verified_profiles/verified_profile_directory.dart b/mobile/lib/shared/verified_profiles/verified_profile_directory.dart index 2f38fd17f7..fa4c9d4696 100644 --- a/mobile/lib/shared/verified_profiles/verified_profile_directory.dart +++ b/mobile/lib/shared/verified_profiles/verified_profile_directory.dart @@ -21,9 +21,18 @@ const _discoveryRefresh = Duration(minutes: 5); /// tolerated, a flood before EOSE fails the batch instead of buffering it. const _maxEarlyPerSubject = 4; +/// Batches smaller than [verifiedProfileBatchSize] allowed before the smallest +/// are merged into the next new one until it fills a batch. +const _maxPartialBatches = 4; + class _Batch { - _Batch(this.subjects); + _Batch(this.subjects, this.interest); final List subjects; + + /// The newest interest when the batch opened. A subject whose interest is + /// newer was evicted and tracked again since, which dropped its `_checked` + /// mark and maybe its cached head, so the batch must be read again. + final int interest; final List early = []; /// Live events awaiting proof, newest per subject, so a fast stream cannot @@ -280,38 +289,84 @@ class VerifiedProfilesNotifier extends Notifier> { _publish(); return; } - final subjects = _wanted.keys.toList(); - final next = >{}; - for (var i = 0; i < subjects.length; i += verifiedProfileBatchSize) { - final end = i + verifiedProfileBatchSize; - final chunk = subjects.sublist( - i, - end < subjects.length ? end : subjects.length, - ); - next[chunk.join(':')] = chunk; - } + // Keep every open batch whose subjects are all still tracked since it + // opened, so newly tracked people get a batch of their own instead of + // re-reading everyone already checked. A retried batch is replaced so a + // fresh snapshot can prove absence, and a batch that lost a subject to + // eviction is replaced too, even if the subject was tracked again. for (final id in _batches.keys.toList()) { - // A retried batch is replaced so a fresh snapshot can prove absence. - if (next.containsKey(id) && !_batches[id]!.stale) continue; - final batch = _batches.remove(id)!..stop(); - _checked.removeAll(batch.subjects); - } - for (final MapEntry(key: id, value: chunk) in next.entries) { - if (!_batches.containsKey(id)) { - unawaited(_open(id, chunk, authority, generation)); + final batch = _batches[id]!; + if (!batch.stale && + batch.subjects.every( + (subject) => (_wanted[subject] ?? 1 << 62) <= batch.interest, + )) { + continue; } + _close(id); + } + var uncovered = _uncoveredSubjects(); + if (uncovered.isEmpty) { + if (_refreshTimer == null) _armRefresh(generation); + _publish(); + return; + } + // Small additions each open a small batch. Past a few of those, merge the + // smallest into the new subjects until they fill a batch, so the + // subscription count stays near the minimum while re-reading as few + // checked subjects as possible. + final partial = [ + for (final batch in _batches.entries) + if (batch.value.subjects.length < verifiedProfileBatchSize) batch, + ]; + if (partial.length >= _maxPartialBatches) { + partial.sort( + (a, b) => a.value.subjects.length.compareTo(b.value.subjects.length), + ); + // Retire at least one unless the new subjects split evenly, since + // they can add a partial batch of their own. + var size = uncovered.length; + var retired = size % verifiedProfileBatchSize == 0; + for (final MapEntry(:key, :value) in partial) { + if (retired && size >= verifiedProfileBatchSize) break; + size += value.subjects.length; + retired = true; + _close(key); + } + uncovered = _uncoveredSubjects(); + } + for (var i = 0; i < uncovered.length; i += verifiedProfileBatchSize) { + final end = i + verifiedProfileBatchSize; + final chunk = uncovered.sublist( + i, + end < uncovered.length ? end : uncovered.length, + ); + unawaited(_open(chunk.join(':'), chunk, authority, generation)); } if (_refreshTimer == null) _armRefresh(generation); _publish(); } + void _close(String id) { + final batch = _batches.remove(id)!..stop(); + _checked.removeAll(batch.subjects); + } + + /// Tracked subjects no open batch covers, oldest interest first. + List _uncoveredSubjects() { + final covered = {for (final batch in _batches.values) ...batch.subjects}; + return [ + for (final key in _wanted.keys) + if (!covered.contains(key)) key, + ]; + } + Future _open( String id, List chunk, VerifiedProfileAuthority authority, int generation, ) async { - final batch = _Batch(chunk); + final batch = _Batch(chunk, _interest); _batches[id] = batch; void fail() { if (!batch.active) return; diff --git a/mobile/test/shared/verified_profiles/verified_profile_directory_test.dart b/mobile/test/shared/verified_profiles/verified_profile_directory_test.dart index ce93027eba..14e878fdc8 100644 --- a/mobile/test/shared/verified_profiles/verified_profile_directory_test.dart +++ b/mobile/test/shared/verified_profiles/verified_profile_directory_test.dart @@ -158,8 +158,7 @@ void main() { expect(h.profiles[subjectA]?.checked, isTrue); h.container.read(verifiedProfilesProvider.notifier).track([subjectB]); h.settle(); - // Rebatching unchecks and rechecks A; every publish rebuilds the map, - // but each head is validated once. + // Every publish rebuilds the map, but each head is validated once. expect(h.profiles[subjectA]?.checked, isTrue); expect(h.profiles[subjectB]?.checked, isTrue); expect(parses..sort(), [subjectA, subjectB]); @@ -202,6 +201,122 @@ void main() { expect(sizes..sort(), [2, 128]); }); + journey('a newly tracked subject is read alone; checked ones stay checked', ( + h, + ) { + h.relay.seed([ + signedProjection(profileTags(subjectA)), + signedProjection(profileTags(subjectB, name: 'Bob', username: 'bob')), + ]); + h.start([subjectA]); + expect(h.profiles[subjectA]?.checked, isTrue); + final before = h.dFilters.length; + final notifier = h.container.read(verifiedProfilesProvider.notifier); + notifier.track([subjectB]); + // A is never unchecked while B's batch opens. + expect(h.profiles[subjectA]?.checked, isTrue); + h.settle(); + expect( + [for (final f in h.dFilters.skip(before)) f['#d']], + [ + [subjectB], + ], + ); + expect(h.profiles[subjectA]?.checked, isTrue); + expect(h.profiles[subjectB]?.checked, isTrue); + }); + + journey('past four small batches, the smallest merge until one is full', (h) { + String key(int i) => i.toRadixString(16).padLeft(64, '0'); + final notifier = h.container.read(verifiedProfilesProvider.notifier); + h.start([for (var i = 0; i < 64; i++) key(i)]); + for (var group = 1; group < 4; group++) { + notifier.track([for (var i = 0; i < 64; i++) key(group * 64 + i)]); + h.settle(); + } + final before = h.dFilters.length; + expect(before, 4, reason: 'four partial batches of 64'); + notifier.track([key(256)]); + h.settle(); + // Two of the 64s fill one batch with the newcomer; the other two stay. + final reads = [ + for (final f in h.dFilters.skip(before)) (f['#d'] as List).cast(), + ]; + expect([for (final r in reads) r.length], [128, 1]); + expect(reads.expand((r) => r), contains(key(256))); + }); + + journey('a large addition past four small batches still retires one', (h) { + String key(int i) => i.toRadixString(16).padLeft(64, '0'); + final notifier = h.container.read(verifiedProfilesProvider.notifier); + h.start([key(0)]); + for (var i = 1; i < 4; i++) { + notifier.track([key(i)]); + h.settle(); + } + final before = h.dFilters.length; + notifier.track([for (var i = 4; i < 133; i++) key(i)]); + h.settle(); + // 129 newcomers would leave a fifth partial; one single folds in. + expect( + [for (final f in h.dFilters.skip(before)) (f['#d'] as List).length], + [128, 2], + ); + }); + + journey('eviction re-reads only the batch that lost a subject', (h) { + String key(int i) => i.toRadixString(16).padLeft(64, '0'); + h.start([for (var i = 0; i < 1024; i++) key(i)]); + final before = h.dFilters.length; + expect(before, 8); + h.container.read(verifiedProfilesProvider.notifier).track([key(1024)]); + h.settle(); + expect( + [for (final f in h.dFilters.skip(before)) f['#d']], + [ + [for (var i = 1; i < 128; i++) key(i), key(1024)], + ], + ); + }); + + journey('a subject evicted and tracked again before the sync is re-read', ( + h, + ) { + String key(int i) => i.toRadixString(16).padLeft(64, '0'); + h.relay.seed([signedProjection(profileTags(subjectA))]); + h.start([subjectA]); + final notifier = h.container.read(verifiedProfilesProvider.notifier); + notifier.track([for (var i = 0; i < 1023; i++) key(i)]); + h.settle(); + expect(h.profiles[subjectA]?.checked, isTrue); + // A is evicted, then tracked again, which evicts key(0) in another batch. + notifier + ..track([key(1023)]) + ..track([subjectA]); + h.settle(); + expect(h.profiles[subjectA]?.checked, isTrue); + }); + + journey('a re-tracked subject keeps its name when the head cache trims', (h) { + String key(int i) => i.toRadixString(16).padLeft(64, '0'); + h.relay.seed([signedProjection(profileTags(subjectA))]); + h.start([subjectA]); + final notifier = h.container.read(verifiedProfilesProvider.notifier); + // Each round evicts A, may trim its cached head, then tracks it again + // before the sync; past 4,096 cached heads only a re-read restores it. + for (var round = 0; round < 6; round++) { + final subjects = [for (var i = 0; i < 1023; i++) key(round * 1023 + i)]; + h.relay.seed([ + for (final s in subjects) signedProjection(profileTags(s)), + ]); + notifier + ..track(subjects) + ..track([subjectA]); + h.settle(); + expect(h.profiles[subjectA]?.displayName, 'Ada Lovelace'); + } + }); + journey('a large roster keeps the 1,024 most recently tracked subjects', (h) { String key(int i) => i.toRadixString(16).padLeft(64, '0'); h.start([for (var i = 0; i < 10; i++) key(i)]);