fix(cache): address review findings — subscription churn, O(n) size, atomicity

P1 fixes from compound engineering review:
- Fix subscription churn flooding relays: DisposableEffect keyed on
  feedMode/noteId (stable) instead of feedState (changes every 250ms)
- BoundedLargeCache: AtomicInteger counter replaces O(n)
  ConcurrentSkipListMap.size() — eliminates 50K traversal per event
- Non-atomic updateHealth(): use MutableStateFlow.update{} for
  atomic compare-and-set instead of read-modify-write race

P2 fixes:
- DesktopThreadFilter: LinkedHashSet for O(1) containment checks,
  down from O(R²) with MutableList
- consumeContactList: createdAt guard ensures newer events always win
- consumeEvent error logging: include event ID and relay URL

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
nrobi144
2026-03-23 09:16:51 +02:00
parent c598c6135c
commit 4f5665a2bd
6 changed files with 75 additions and 50 deletions
@@ -22,24 +22,22 @@ package com.vitorpamplona.amethyst.desktop.cache
import com.vitorpamplona.quartz.utils.cache.CacheCollectors import com.vitorpamplona.quartz.utils.cache.CacheCollectors
import com.vitorpamplona.quartz.utils.cache.LargeCache import com.vitorpamplona.quartz.utils.cache.LargeCache
import java.util.concurrent.atomic.AtomicInteger
/** /**
* A bounded wrapper around [LargeCache] that enforces a maximum size. * A bounded wrapper around [LargeCache] that enforces a maximum size.
* *
* When the cache exceeds [maxSize], the oldest entries (by key order) are evicted. * When the cache exceeds [maxSize], entries are evicted by key order.
* Uses [LargeCache] (ConcurrentSkipListMap) for lock-free reads and rich query APIs * Uses [LargeCache] (ConcurrentSkipListMap) for lock-free reads and rich query APIs.
* (filterIntoSet, mapNotNull, etc.) matching Android's LocalCache patterns.
* *
* Chosen over LruCache because: * Size tracking uses AtomicInteger (O(1)) instead of ConcurrentSkipListMap.size() (O(n)).
* - Lock-free reads via ConcurrentSkipListMap (vs synchronized on every get())
* - Rich query API matching Android's filter patterns
* - No snapshot copy overhead for iteration
*/ */
class BoundedLargeCache<K : Comparable<K>, V>( class BoundedLargeCache<K : Comparable<K>, V>(
private val maxSize: Int, private val maxSize: Int,
private val evictPercent: Float = 0.1f, private val evictPercent: Float = 0.1f,
) { ) {
private val inner = LargeCache<K, V>() private val inner = LargeCache<K, V>()
private val sizeCounter = AtomicInteger(0)
fun get(key: K): V? = inner.get(key) fun get(key: K): V? = inner.get(key)
@@ -47,7 +45,9 @@ class BoundedLargeCache<K : Comparable<K>, V>(
key: K, key: K,
value: V, value: V,
) { ) {
val existing = inner.get(key)
inner.put(key, value) inner.put(key, value)
if (existing == null) sizeCounter.incrementAndGet()
enforceSize() enforceSize()
} }
@@ -55,20 +55,33 @@ class BoundedLargeCache<K : Comparable<K>, V>(
key: K, key: K,
builder: (K) -> V, builder: (K) -> V,
): V { ): V {
val existing = inner.get(key)
if (existing != null) return existing
val result = inner.getOrCreate(key, builder) val result = inner.getOrCreate(key, builder)
// Increment if we were the ones who created it (not a concurrent insert)
if (inner.get(key) === result) {
sizeCounter.incrementAndGet()
}
enforceSize() enforceSize()
return result return result
} }
fun remove(key: K): V? = inner.remove(key) fun remove(key: K): V? {
val removed = inner.remove(key)
if (removed != null) sizeCounter.decrementAndGet()
return removed
}
fun containsKey(key: K): Boolean = inner.containsKey(key) fun containsKey(key: K): Boolean = inner.containsKey(key)
fun size(): Int = inner.size() fun size(): Int = sizeCounter.get()
fun isEmpty(): Boolean = inner.isEmpty() fun isEmpty(): Boolean = sizeCounter.get() == 0
fun clear() = inner.clear() fun clear() {
inner.clear()
sizeCounter.set(0)
}
fun keys(): Set<K> = inner.keys() fun keys(): Set<K> = inner.keys()
@@ -83,11 +96,15 @@ class BoundedLargeCache<K : Comparable<K>, V>(
fun count(consumer: CacheCollectors.BiFilter<K, V>): Int = inner.count(consumer) fun count(consumer: CacheCollectors.BiFilter<K, V>): Int = inner.count(consumer)
private fun enforceSize() { private fun enforceSize() {
val currentSize = inner.size() val currentSize = sizeCounter.get()
if (currentSize > maxSize) { if (currentSize > maxSize) {
val toRemove = (maxSize * evictPercent).toInt().coerceAtLeast(1) val toRemove = (maxSize * evictPercent).toInt().coerceAtLeast(1)
val keys = inner.keys().take(toRemove) val keys = inner.keys().take(toRemove)
keys.forEach { inner.remove(it) } keys.forEach {
if (inner.remove(it) != null) {
sizeCounter.decrementAndGet()
}
}
} }
} }
} }
@@ -295,12 +295,13 @@ class DesktopLocalCache : ICacheProvider {
* Consumes a kind 3 contact list event (replaceable). * Consumes a kind 3 contact list event (replaceable).
* Updates the cached followedUsers set. * Updates the cached followedUsers set.
*/ */
private var lastContactListCreatedAt = 0L
private fun consumeContactList(event: ContactListEvent): Boolean { private fun consumeContactList(event: ContactListEvent): Boolean {
val currentFollows = _followedUsers.value // Replaceable event — only accept newer contact lists
val newFollows = event.verifiedFollowKeySet() if (event.createdAt <= lastContactListCreatedAt) return false
if (newFollows != currentFollows) { lastContactListCreatedAt = event.createdAt
_followedUsers.value = newFollows _followedUsers.value = event.verifiedFollowKeySet()
}
return true return true
} }
@@ -94,19 +94,20 @@ class DesktopThreadFilter(
override fun feed(): List<Note> { override fun feed(): List<Note> {
val root = cache.getNoteIfExists(noteId) ?: return emptyList() val root = cache.getNoteIfExists(noteId) ?: return emptyList()
val result = mutableListOf(root) // Use LinkedHashSet for O(1) containment checks (was O(R) with MutableList)
collectReplies(root, result) val seen = LinkedHashSet<Note>()
return result.sortedWith(compareBy { it.createdAt() ?: 0 }) seen.add(root)
collectReplies(root, seen)
return seen.sortedWith(compareBy { it.createdAt() ?: 0 })
} }
private fun collectReplies( private fun collectReplies(
note: Note, note: Note,
result: MutableList<Note>, seen: LinkedHashSet<Note>,
) { ) {
for (reply in note.replies) { for (reply in note.replies) {
if (reply !in result) { if (seen.add(reply)) {
result.add(reply) collectReplies(reply, seen)
collectReplies(reply, result)
} }
} }
} }
@@ -40,6 +40,7 @@ import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
@@ -120,15 +121,16 @@ class DesktopRelaySubscriptionsCoordinator(
lastEventReceivedAt: Long? = null, lastEventReceivedAt: Long? = null,
eoseReceived: Boolean? = null, eoseReceived: Boolean? = null,
) { ) {
_subscriptionHealth.value = _subscriptionHealth.update { current ->
_subscriptionHealth.value.toMutableMap().apply { current.toMutableMap().apply {
val current = this[subId] ?: SubscriptionHealth() val existing = this[subId] ?: SubscriptionHealth()
this[subId] = this[subId] =
current.copy( existing.copy(
lastEventReceivedAt = lastEventReceivedAt ?: current.lastEventReceivedAt, lastEventReceivedAt = lastEventReceivedAt ?: existing.lastEventReceivedAt,
eoseReceived = eoseReceived ?: current.eoseReceived, eoseReceived = eoseReceived ?: existing.eoseReceived,
) )
} }
}
} }
/** /**
@@ -150,7 +152,7 @@ class DesktopRelaySubscriptionsCoordinator(
} }
} }
} catch (e: Exception) { } catch (e: Exception) {
println("Coordinator: failed to consume kind ${event.kind}: ${e.message}") println("Coordinator: failed to consume kind=${event.kind} id=${event.id} relay=$relay: ${e.message}")
} }
} }
} }
@@ -230,8 +232,7 @@ class DesktopRelaySubscriptionsCoordinator(
fun releaseInteractions(subId: String) { fun releaseInteractions(subId: String) {
screenSubscriptions.remove(subId)?.cancel() screenSubscriptions.remove(subId)?.cancel()
client.close(subId) client.close(subId)
_subscriptionHealth.value = _subscriptionHealth.update { it - subId }
_subscriptionHealth.value.toMutableMap().apply { remove(subId) }
} }
/** /**
@@ -204,17 +204,19 @@ fun FeedScreen(
} }
} }
// Request interaction subscriptions (zaps, reactions, reposts) for visible notes // Request interaction subscriptions — keyed on feedMode (stable), not feedState (changes every 250ms)
DisposableEffect(feedState, subscriptionsCoordinator) { DisposableEffect(feedMode, subscriptionsCoordinator) {
val coordinator = subscriptionsCoordinator ?: return@DisposableEffect onDispose {} val coordinator = subscriptionsCoordinator ?: return@DisposableEffect onDispose {}
val notes = viewModel.feedState.visibleNotes() val relays = relayManager.relayStatuses.value.keys
val noteIds = notes.mapNotNull { it.event?.id } // Initial subscription with whatever notes are visible now
if (noteIds.isEmpty()) return@DisposableEffect onDispose {} val noteIds = viewModel.feedState.visibleNotes().mapNotNull { it.event?.id }
val subId =
val relays = if (noteIds.isNotEmpty()) {
relayManager.relayStatuses.value.keys coordinator.requestInteractions(noteIds, relays)
val subId = coordinator.requestInteractions(noteIds, relays) } else {
onDispose { coordinator.releaseInteractions(subId) } null
}
onDispose { subId?.let { coordinator.releaseInteractions(it) } }
} }
@OptIn(ExperimentalLayoutApi::class) @OptIn(ExperimentalLayoutApi::class)
@@ -155,15 +155,18 @@ fun ThreadScreen(
} }
} }
// Request interaction data (zaps, reactions, reposts) for visible thread notes // Request interaction data — keyed on noteId (stable), not threadNotes (changes on every bundle)
DisposableEffect(threadNotes, subscriptionsCoordinator) { DisposableEffect(noteId, subscriptionsCoordinator) {
val coordinator = subscriptionsCoordinator ?: return@DisposableEffect onDispose {} val coordinator = subscriptionsCoordinator ?: return@DisposableEffect onDispose {}
val noteIds = threadNotes.mapNotNull { it.event?.id } val noteIds = threadNotes.mapNotNull { it.event?.id }
if (noteIds.isEmpty()) return@DisposableEffect onDispose {}
val relays = relayManager.relayStatuses.value.keys val relays = relayManager.relayStatuses.value.keys
val subId = coordinator.requestInteractions(noteIds, relays) val subId =
onDispose { coordinator.releaseInteractions(subId) } if (noteIds.isNotEmpty()) {
coordinator.requestInteractions(noteIds, relays)
} else {
null
}
onDispose { subId?.let { coordinator.releaseInteractions(it) } }
} }
// Load metadata for thread authors via coordinator // Load metadata for thread authors via coordinator