synchronize all cache mutations and supply sinceRelaySet so callers get snapshot copies

rework groupByRelayPresence to build relay snapshots via sinceRelaySet,
    filtering with immutable lists to prevent ConcurrentModificationException when relays update mid-iteration
This commit is contained in:
davotoula
2025-11-05 19:02:53 +01:00
parent a82d6565fa
commit 545cd2ff6d
2 changed files with 24 additions and 8 deletions
@@ -90,19 +90,30 @@ class UserReportsSubAssembler(
users: Iterable<User>,
eoseCache: EOSEAccountFast<User>,
inRelays: Set<NormalizedRelayUrl>,
): Collection<List<User>> =
users
.groupBy {
eoseCache
.since(it)
?.keys
?.intersect(inRelays)
?.hashCode()
): Collection<List<User>> {
if (users.none()) return emptyList()
val relaySnapshot = inRelays.toSet()
return users
.groupBy { user ->
val relaysForUser = eoseCache.sinceRelaySet(user)
if (relaysForUser.isNullOrEmpty() || relaySnapshot.isEmpty()) {
null
} else {
val intersection = relaysForUser.filter { it in relaySnapshot }.sorted()
if (intersection.isEmpty()) {
null
} else {
intersection.hashCode()
}
}
}.values
.map {
// important to keep in order otherwise the Relay thinks the filter has changed and we REQ again
it.sortedBy { it.pubkeyHex }
}
}
fun findMinimumEOSEsForUsers(
users: List<User>,
@@ -166,6 +166,11 @@ class EOSEAccountFast<T : Any>(
users[key]?.relayList?.toMutableMap()
}
fun sinceRelaySet(key: T): Set<NormalizedRelayUrl>? =
synchronized(lock) {
users[key]?.relayList?.keys?.toSet()
}
fun newEose(
user: T,
relayUrl: NormalizedRelayUrl,