perf(mobile): keep verified-name batches open as people are added (#9450)

Verified names are read in subscriptions of up to 128 people. Each time
someone new was tracked, the batches were rebuilt from the whole tracked
list, so the last partial batch got a new identity, was closed, and was
reopened with everyone in it plus the newcomer. Profiles show up a few
at a time (search results, channel members, mentions), so the same
records were read again and again. Past 1,024 tracked people, each
newcomer evicts the oldest, which shifts every batch boundary, so every
batch was closed and all 1,024 records were read again.

Now an open batch stays open as long as all its people are still
tracked. Newly tracked people get a batch of their own. A batch that
loses someone to eviction, or that the relay retried, is replaced as
before, and so is a batch whose person was evicted and tracked again
before the batch was re-checked, so their check mark and cached name are
re-read. Once four batches are smaller than 128, the next addition
merges the smallest of them with the newcomers until they fill a batch
(at least one, so no fifth small batch appears). The subscription count
stays within four of the minimum, and a merge re-reads at most 128 plus
the largest small batch.

People already checked also stay checked while a newcomer's batch opens;
before, the reopened batch briefly unchecked them.

### Validation

New tests: tracking one more person opens a batch with only that person,
and the people already checked stay checked throughout; with four small
batches of 64 open, one more person merges two of them into a full batch
and leaves the other two open; a large addition past four small batches
still retires one; tracking one more person at the 1,024 limit re-reads
only the batch that lost the oldest person; a person evicted and tracked
again before the sync is re-read and checked; and a re-tracked person
keeps their verified name when the cached-name limit trims it. Existing
batch, eviction, retry and reconnect tests pass unchanged (one comment
updated). Full mobile test suite and `flutter analyze` pass.

On an iPhone 12 mini signed in to Block's Builderlab (instrumented
profile build of e2fb02c), opening search and typing "presence" after
opening a channel and a thread: verified-name reads went from 21
subscriptions and 328 KB before this change to 11 and 88 KB with it
(counted per subscription opened; the first revision measured the same
way gave 11 and 90 KB).

The same automated run covered cold start, opening the-squad and one of
its threads, search with the People filter, opening buzz-mobile's
details and scrolling its full 86-person member list, and a 60 s
background then resume (the resync re-read 2 batches, 65 KB). Names
rendered throughout.

### Regressions

Up to four more open subscriptions while small batches accumulate, each
held open by the relay like the existing ones. A community with under
128 tracked people now has up to five instead of one. No merge policy
reads less than the old slicing on every sequence of additions. In a
model of 3,000 random tracking sequences, this one read more records
than the old code in about 1 sequence in 500, by at most a quarter.

---------

Signed-off-by: Tom Brow <tomb@squareup.com>
Co-authored-by: Tom Brow <tomb@squareup.com>
Co-authored-by: studio <dc2fc481343672df31a689840c09299972e704dac592e0da6f18486646a7ee3e@buzz.block.builderlab.xyz>
Co-authored-by: studio <dc2fc481343672df31a689840c09299972e704dac592e0da6f18486646a7ee3e@localhost>

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<String> 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<NostrEvent> early = [];
 
   /// Live events awaiting proof, newest per subject, so a fast stream cannot
@@ -280,38 +289,84 @@ class VerifiedProfilesNotifier extends Notifier<Map<String, VerifiedProfile>> {
       _publish();
       return;
     }
-    final subjects = _wanted.keys.toList();
-    final next = <String, List<String>>{};
-    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<String> _uncoveredSubjects() {
+    final covered = {for (final batch in _batches.values) ...batch.subjects};
+    return [
+      for (final key in _wanted.keys)
+        if (!covered.contains(key)) key,
+    ];
+  }
+
   Future<void> _open(
     String id,
     List<String> chunk,
     VerifiedProfileAuthority authority,
     int generation,
   ) async {
-    final batch = _Batch(chunk);
+    final batch = _Batch(chunk, _interest);
     _batches[id] = batch;
     void fail() {
       if (!batch.active) return;

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<String>(),
+    ];
+    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)]);