fix: record DVM heartbeats in a strong registry — beat notes are WeakReference-held

Beat notes live in LocalCache.addressables (WeakReference values) with no
strong holder on the Discover screen: every GC sweep cleared them all at
once, the freshness gate failed for every DVM simultaneously, and the list
collapsed and rebuilt one beat at a time (worst on refresh — an allocation
spike). The detail screen survived only because its own composable held the
note.

DvmHeartbeatRegistry (Address -> latest createdAt, strong, one entry per
DVM) is now the freshness source: recorded on every beat consumption, read
by the gate and the liveness composables. Stale entries are inert; the map
is bounded by distinct DVM addresses.
This commit is contained in:
Dr. Tobias Baur
2026-09-10 15:49:55 +02:00
parent 9db245f609
commit f8040e5962
5 changed files with 95 additions and 4 deletions
@@ -30,11 +30,18 @@ import com.vitorpamplona.quartz.utils.TimeUtils
/** The cache slot a DVM's heartbeat lives in: the announcement's own address, kind 11998. */ /** The cache slot a DVM's heartbeat lives in: the announcement's own address, kind 11998. */
fun LocalCache.dvmHeartbeatOf(appDef: AppDefinitionEvent): DvmHeartbeatEvent? = getAddressableNoteIfExists(Address(DvmHeartbeatEvent.KIND, appDef.pubKey, appDef.dTag()))?.event as? DvmHeartbeatEvent fun LocalCache.dvmHeartbeatOf(appDef: AppDefinitionEvent): DvmHeartbeatEvent? = getAddressableNoteIfExists(Address(DvmHeartbeatEvent.KIND, appDef.pubKey, appDef.dTag()))?.event as? DvmHeartbeatEvent
/** A DVM counts as alive only if its latest heartbeat is at most 420s old. */ /**
* A DVM counts as alive only if its latest heartbeat is at most 900s old. The registry (not the
* WeakReference-held beat note) is the freshness source: beat notes have no strong holder on the
* Discover screen, and a GC sweep cleared them all at once, collapsing the list.
*/
fun LocalCache.hasFreshDvmHeartbeat( fun LocalCache.hasFreshDvmHeartbeat(
appDef: AppDefinitionEvent, appDef: AppDefinitionEvent,
now: Long = TimeUtils.now(), now: Long = TimeUtils.now(),
): Boolean = dvmHeartbeatOf(appDef)?.isFreshAt(now) == true ): Boolean =
DvmHeartbeatRegistry
.latestAt(Address(DvmHeartbeatEvent.KIND, appDef.pubKey, appDef.dTag()))
?.let { it >= now - DvmHeartbeatEvent.MAX_AGE_SECONDS } == true
/** /**
* Every cached content-discovery announcement, WITHOUT the freshness gate — this is the source the * Every cached content-discovery announcement, WITHOUT the freshness gate — this is the source the
@@ -0,0 +1,58 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.amethyst.model
import com.vitorpamplona.quartz.nip01Core.core.Address
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.update
import java.util.concurrent.ConcurrentHashMap
/**
* Strong, process-wide record of the latest heartbeat per DVM announcement address
* (`Address(11998, dvmPubKey, dTag) -> createdAt`).
*
* Beat Notes themselves live in `LocalCache.addressables`, a WeakReference store with no strong
* holder on the Discover screen — every GC sweep cleared them all at once and the freshness gate
* collapsed for every DVM simultaneously (the list emptied and rebuilt one beat at a time). This
* registry is the freshness source the gate and the liveness composables read: strong references,
* fed by every beat-arrival path (global REQ, outbox batches, per-surface fetches — all beat
* consumption routes through [record]).
*
* One entry per DVM address ever seen; timestamps only, so it stays tiny. `0` means "no beat".
*/
object DvmHeartbeatRegistry {
private val latestBeatCreatedAt = ConcurrentHashMap<Address, MutableStateFlow<Long>>()
private fun flowFor(address: Address): MutableStateFlow<Long> = latestBeatCreatedAt.getOrPut(address) { MutableStateFlow(0L) }
/** Observable latest-beat timestamp for this address; `0` means "no beat seen yet". */
fun flowForPublic(address: Address): StateFlow<Long> = flowFor(address)
/** Records a beat's createdAt; older beats never move the entry backwards. */
fun record(
address: Address,
createdAt: Long,
) = flowFor(address).update { current -> if (createdAt > current) createdAt else current }
/** The latest recorded beat's createdAt for this address, or null when no beat was ever seen. */
fun latestAt(address: Address): Long? = flowFor(address).value.takeIf { it > 0L }
}
@@ -3744,8 +3744,14 @@ object LocalCache : ILocalCache, ICacheProvider, Dao {
is CommunityListEvent, is CommunityListEvent,
is ContactListEvent, is ContactListEvent,
// DVM heartbeat (11998): stored per Address(11998, author, d) so liveness checks find the // DVM heartbeat (11998): stored per Address(11998, author, d) so liveness checks find the
// beat at the announcement's mirror address (amethyst/plans/2026-09-10-dvm-heartbeat-liveness.md). // beat at the announcement's mirror address (amethyst/plans/2026-09-10-dvm-heartbeat-liveness.md),
// AND recorded into the strong registry — beat notes are WeakReference-held with no strong
// holder on the Discover screen, so the gate must not depend on them surviving GC.
is DvmHeartbeatEvent, is DvmHeartbeatEvent,
->
consumeBaseReplaceable(event, relay, wasVerified).also {
DvmHeartbeatRegistry.record(event.address(), event.createdAt)
}
is EmojiPackEvent, is EmojiPackEvent,
is EmojiPackSelectionEvent, is EmojiPackSelectionEvent,
is EphemeralChatListEvent, is EphemeralChatListEvent,
@@ -39,8 +39,10 @@ import androidx.compose.runtime.setValue
import androidx.compose.ui.Alignment import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier import androidx.compose.ui.Modifier
import androidx.compose.ui.unit.dp import androidx.compose.ui.unit.dp
import androidx.lifecycle.compose.collectAsStateWithLifecycle
import com.vitorpamplona.amethyst.commons.resources.Res import com.vitorpamplona.amethyst.commons.resources.Res
import com.vitorpamplona.amethyst.commons.resources.dvm_offline_banner import com.vitorpamplona.amethyst.commons.resources.dvm_offline_banner
import com.vitorpamplona.amethyst.model.DvmHeartbeatRegistry
import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.observeNoteAndMap import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.observeNoteAndMap
import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
import com.vitorpamplona.amethyst.ui.stringRes import com.vitorpamplona.amethyst.ui.stringRes
@@ -81,6 +83,11 @@ fun rememberDvmHeartbeatFresh(
val observed = val observed =
heartbeatNote?.let { observeNoteAndMap(it, accountViewModel) { it.event as? DvmHeartbeatEvent } } heartbeatNote?.let { observeNoteAndMap(it, accountViewModel) { it.event as? DvmHeartbeatEvent } }
// The registry is the eviction-proof freshness source (beat notes are WeakReference-held);
// the observed note is the secondary source and also drives the outbox fetch.
val registryBeat by
DvmHeartbeatRegistry.flowForPublic(appDefinitionAddress).collectAsStateWithLifecycle()
var now by remember(heartbeatNote) { mutableLongStateOf(TimeUtils.now()) } var now by remember(heartbeatNote) { mutableLongStateOf(TimeUtils.now()) }
LaunchedEffect(heartbeatNote) { LaunchedEffect(heartbeatNote) {
while (isActive) { while (isActive) {
@@ -90,7 +97,8 @@ fun rememberDvmHeartbeatFresh(
} }
val beat = observed?.value val beat = observed?.value
return rememberUpdatedState(beat != null && beat.isFreshAt(now)) val bestBeat = maxOf(beat?.createdAt ?: 0L, registryBeat)
return rememberUpdatedState(bestBeat > 0L && bestBeat >= now - DvmHeartbeatEvent.MAX_AGE_SECONDS)
} }
/** Floating "DVM is offline" banner, mirroring the Home status banner's card style. */ /** Floating "DVM is offline" banner, mirroring the Home status banner's card style. */
@@ -121,6 +121,18 @@ class DvmHeartbeatTest {
assertFalse(LocalCache.hasFreshDvmHeartbeat(appDef("dvm-never"), TimeUtils.now())) assertFalse(LocalCache.hasFreshDvmHeartbeat(appDef("dvm-never"), TimeUtils.now()))
} }
@Test
fun theGateSurvivesWeakCacheEvictionOfTheBeatNote() {
// Beat notes live in LocalCache's WeakReference store with no strong holder on the
// Discover screen — a GC sweep clears them all at once and the list collapses. The
// freshness gate must therefore read the strong registry, not the evictable note.
val app = appDef("dvm-registry")
DvmHeartbeatRegistry.record(Address(DvmHeartbeatEvent.KIND, appDefPubKey, "dvm-registry"), 1_760_000_000L - 100)
assertTrue("registry alone proves liveness", LocalCache.hasFreshDvmHeartbeat(app, 1_760_000_000L))
assertTrue("no entry for dvm-never", DvmHeartbeatRegistry.latestAt(Address(DvmHeartbeatEvent.KIND, appDefPubKey, "dvm-never")) == null)
}
@Test @Test
fun theUngatedAnnouncementScanKeepsDvmsTheGateWouldHide() { fun theUngatedAnnouncementScanKeepsDvmsTheGateWouldHide() {
// The outbox fetcher must source announcements from the cache, NOT from the gated feed // The outbox fetcher must source announcements from the cache, NOT from the gated feed