nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

More debugging information

Browse Source
This commit is contained in:
Vitor Pamplona
2026-03-19 09:34:52 -04:00
parent 3e3adaff10
commit ec29a47dc4
4 changed files with 179 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt
+110 -66
View File
@@ -38,7 +38,6 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.tags.people.isTaggedUser
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
@@ -50,8 +49,7 @@ import kotlinx.coroutines.launch
import kotlinx.coroutines.supervisorScope
import kotlinx.coroutines.sync.Semaphore
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicLong
import kotlin.coroutines.cancellation.CancellationException
/**
* Syncs the user's events across all known relays:
@@ -103,12 +101,24 @@ class EventSync(
object Idle : SyncState()
data class Running(
val relaysCompleted: Int,
val totalRelays: Int,
val eventsSent: Int,
) : SyncState()
val relaysCompleted: MutableStateFlow<Int>,
val totalRelays: MutableStateFlow<Int>,
val eventsSent: MutableStateFlow<Int>,
val eventsReceived: MutableStateFlow<Int>,
val eventsAccepted: MutableStateFlow<Int>,
) : SyncState() {
constructor(relaysCompleted: Int, totalRelays: Int, eventsSent: Int, eventsReceived: Int, eventsAccepted: Int) :
this(
relaysCompleted = MutableStateFlow(relaysCompleted),
totalRelays = MutableStateFlow(totalRelays),
eventsSent = MutableStateFlow(eventsSent),
eventsReceived = MutableStateFlow(eventsReceived),
eventsAccepted = MutableStateFlow(eventsAccepted),
)
}
data class Done(
val totalEventsReceived: Int,
val totalEventsSent: Int,
val totalEventsAccepted: Int,
val durationMs: Long,
@@ -180,9 +190,16 @@ class EventSync(
val status: MutableStateFlow<ConnectionStatus>,
val eventsFound: MutableStateFlow<Int>,
val eventsAccepted: MutableStateFlow<Int>,
val pageUntil: MutableStateFlow<Long?> = MutableStateFlow(null),
) {
constructor(relay: NormalizedRelayUrl, status: ConnectionStatus, eventsFound: Int, eventsAccepted: Int) :
this(relay, MutableStateFlow(status), MutableStateFlow(eventsFound), MutableStateFlow(eventsAccepted))
constructor(relay: NormalizedRelayUrl, status: ConnectionStatus, eventsFound: Int, eventsAccepted: Int, untilPage: Long? = null) :
this(
relay = relay,
status = MutableStateFlow(status),
eventsFound = MutableStateFlow(eventsFound),
eventsAccepted = MutableStateFlow(eventsAccepted),
pageUntil = MutableStateFlow(untilPage),
)
}
/**
@@ -346,14 +363,17 @@ class EventSync(
val outboxDedup = ConcurrentHashMap.newKeySet<String>()
val inboxDedup = ConcurrentHashMap.newKeySet<String>()
val dmDedup = ConcurrentHashMap.newKeySet<String>()
val totalSent = AtomicLong(0)
// OK (true) tracking: maps each sent event ID to its source relay.
// The first OK true for an event atomically removes it from this map,
// crediting the acceptance to the source relay and preventing double-counting.
val sourceRelayOfEvent = ConcurrentHashMap<HexKey, NormalizedRelayUrl>()
val acceptedCountPerSourceRelay = ConcurrentHashMap<NormalizedRelayUrl, AtomicInteger>()
val totalAccepted = AtomicLong(0)
val runningState =
SyncState.Running(
relaysCompleted = 0,
totalRelays = totalRelays,
eventsSent = 0,
eventsReceived = 0,
eventsAccepted = 0,
)
val okListener =
object : IRelayClientListener {
@@ -375,22 +395,30 @@ class EventSync(
success: Boolean,
) {
super.onSent(relay, cmdStr, cmd, success)
if (cmd is EventCmd) {
var hasSent = false
if (outboxDedup.contains(cmd.event.id)) {
liveActivity.value.outboxTargets[relay.url]
?.eventsSent
?.update { it + 1 }
hasSent = true
}
if (inboxDedup.contains(cmd.event.id)) {
liveActivity.value.inboxTargets[relay.url]
?.eventsSent
?.update { it + 1 }
hasSent = true
}
if (dmDedup.contains(cmd.event.id)) {
liveActivity.value.dmTargets[relay.url]
?.eventsSent
?.update { it + 1 }
hasSent = true
}
if (hasSent) {
runningState.eventsSent.update { it + 1 }
}
} else if (cmd is ReqCmd) {
val currentStatus = liveActivity.value.runningRelays[relay.url]?.status
@@ -407,37 +435,39 @@ class EventSync(
) {
if (msg is OkMessage && msg.success && msg.message.isBlank()) {
// remove() is atomic: returns non-null only for the first OK per event.
val sourceRelay = sourceRelayOfEvent.remove(msg.eventId) ?: return
acceptedCountPerSourceRelay.getOrPut(sourceRelay) { AtomicInteger(0) }.incrementAndGet()
totalAccepted.incrementAndGet()
val sourceRelay = sourceRelayOfEvent.remove(msg.eventId)
if (sourceRelay != null) {
liveActivity.value.runningRelays[sourceRelay]
?.eventsAccepted
?.update { it + 1 }
}
if (outboxDedup.contains(msg.eventId)) {
liveActivity.value.outboxTargets[relay.url]
?.eventsAccepted
?.update { it + 1 }
val relayTarget = liveActivity.value.outboxTargets[relay.url]
if (relayTarget != null) {
relayTarget.eventsAccepted.update { it + 1 }
runningState.eventsAccepted.update { it + 1 }
}
}
if (dmDedup.contains(msg.eventId)) {
liveActivity.value.dmTargets[relay.url]
?.eventsAccepted
?.update { it + 1 }
val relayTarget = liveActivity.value.dmTargets[relay.url]
if (relayTarget != null) {
relayTarget.eventsAccepted.update { it + 1 }
runningState.eventsAccepted.update { it + 1 }
}
}
if (inboxDedup.contains(msg.eventId)) {
liveActivity.value.inboxTargets[relay.url]
?.eventsAccepted
?.update { it + 1 }
val relayTarget = liveActivity.value.inboxTargets[relay.url]
if (relayTarget != null) {
relayTarget.eventsAccepted.update { it + 1 }
runningState.eventsAccepted.update { it + 1 }
}
}
}
}
}
val relaysCompleted = AtomicInteger(0)
_syncState.value =
SyncState.Running(
relaysCompleted = relaysCompleted.get(),
totalRelays = totalRelays,
eventsSent = totalSent.get().toInt(),
)
_syncState.emit(runningState)
clientBuilder().use { client ->
client.subscribe(okListener)
@@ -445,6 +475,11 @@ class EventSync(
client.downloadFromPool(
relays = relaysToProcess,
filters = perRelayFilters,
onNewPage = { until, sourceRelay ->
_liveActivity.value.runningRelays[sourceRelay]
?.pageUntil
?.tryEmit(until)
},
onEvent = { event, sourceRelay ->
val isMyEvent = event.pubKey == myPubKey
val mentionsMe = event.tags.isTaggedUser(myPubKey)
@@ -452,36 +487,41 @@ class EventSync(
val live = liveActivity.value
var newEvent = false
var matchesAtLeastOneFilter = false
// Each routing rule is independent: an event can match more than one.
if (isMyEvent && outboxTargets.isNotEmpty()) {
if (outboxDedup.add(event.id)) {
sourceRelayOfEvent[event.id] = sourceRelay
client.send(event, outboxTargets)
totalSent.incrementAndGet()
live.runningRelays[sourceRelay]?.eventsFound?.update { it + 1 }
live.completedRelays[sourceRelay]?.eventsFound?.update { it + 1 }
newEvent = true
}
matchesAtLeastOneFilter = true
}
if (mentionsMe && isDmKind && dmTargets.isNotEmpty()) {
if (dmDedup.add(event.id)) {
sourceRelayOfEvent[event.id] = sourceRelay
client.send(event, dmTargets)
totalSent.incrementAndGet()
live.runningRelays[sourceRelay]?.eventsFound?.update { it + 1 }
live.completedRelays[sourceRelay]?.eventsFound?.update { it + 1 }
newEvent = true
}
matchesAtLeastOneFilter = true
}
if (mentionsMe && !isDmKind && inboxTargets.isNotEmpty()) {
if (inboxDedup.add(event.id)) {
sourceRelayOfEvent[event.id] = sourceRelay
client.send(event, inboxTargets)
totalSent.incrementAndGet()
live.runningRelays[sourceRelay]?.eventsFound?.update { it + 1 }
live.completedRelays[sourceRelay]?.eventsFound?.update { it + 1 }
newEvent = true
}
matchesAtLeastOneFilter = true
}
if (newEvent) {
sourceRelayOfEvent[event.id] = sourceRelay
}
if (matchesAtLeastOneFilter) {
runningState.eventsReceived.update { it + 1 }
live.runningRelays[sourceRelay]?.eventsFound?.update { it + 1 }
live.completedRelays[sourceRelay]?.eventsFound?.update { it + 1 }
}
},
onRelayStart = { relay ->
@@ -513,30 +553,27 @@ class EventSync(
status?.tryEmit(LiveSyncActivity.ConnectionStatus.Completed)
}
_syncState.value =
SyncState.Running(
relaysCompleted = relaysCompleted.incrementAndGet(),
totalRelays = totalRelays,
eventsSent = totalSent.get().toInt(),
)
runningState.relaysCompleted.update { it + 1 }
},
)
_syncState.value =
SyncState.Done(
totalEventsSent = totalSent.get().toInt(),
totalEventsAccepted = totalAccepted.get().toInt(),
totalEventsReceived = runningState.eventsReceived.value,
totalEventsSent = runningState.eventsSent.value,
totalEventsAccepted = runningState.eventsAccepted.value,
durationMs = System.currentTimeMillis() - startTime,
)
} catch (e: CancellationException) {
} catch (e: Exception) {
_syncState.value =
SyncState.Done(
totalEventsSent = totalSent.get().toInt(),
totalEventsAccepted = totalAccepted.get().toInt(),
totalEventsReceived = runningState.eventsReceived.value,
totalEventsSent = runningState.eventsSent.value,
totalEventsAccepted = runningState.eventsAccepted.value,
durationMs = System.currentTimeMillis() - startTime,
)
throw e
} catch (e: Exception) {
if (e is CancellationException) throw e
_syncState.value = SyncState.Error(e.message ?: "Unknown error")
} finally {
client.unsubscribe(okListener)
@@ -554,6 +591,7 @@ class EventSync(
private suspend fun INostrClient.downloadFromPool(
relays: List<NormalizedRelayUrl>,
filters: Map<NormalizedRelayUrl, List<Filter>>,
onNewPage: (Long, NormalizedRelayUrl) -> Unit,
onEvent: (Event, NormalizedRelayUrl) -> Unit,
onRelayStart: (NormalizedRelayUrl) -> Unit,
onRelayComplete: (NormalizedRelayUrl) -> Unit,
@@ -567,7 +605,12 @@ class EventSync(
try {
onRelayStart(relay)
filters[relay]?.let { filtersForRelay ->
downloadFromRelay(relay, filtersForRelay) { event -> onEvent(event, relay) }
downloadFromRelay(
relay = relay,
filters = filtersForRelay,
onNewPage = { onNewPage(it, relay) },
onEvent = { onEvent(it, relay) },
)
} ?: 0
onRelayComplete(relay)
} finally {
@@ -587,6 +630,7 @@ class EventSync(
private suspend fun INostrClient.downloadFromRelay(
relay: NormalizedRelayUrl,
filters: List<Filter>,
onNewPage: (Long) -> Unit,
onEvent: (Event) -> Unit,
): Int = reqBypassingRelayLimits(relay, filters, RELAY_TIMEOUT_MS, onEvent)
): Int = reqBypassingRelayLimits(relay, filters, RELAY_TIMEOUT_MS, onNewPage, onEvent)
}
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt
+60 -26
View File
@@ -54,6 +54,7 @@ import androidx.compose.runtime.setValue
import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier
import androidx.compose.ui.draw.clip
import androidx.compose.ui.platform.LocalContext
import androidx.compose.ui.text.font.FontWeight
import androidx.compose.ui.text.style.TextAlign
import androidx.compose.ui.text.style.TextOverflow
@@ -63,6 +64,7 @@ import androidx.lifecycle.compose.collectAsStateWithLifecycle
import com.vitorpamplona.amethyst.R
import com.vitorpamplona.amethyst.ui.navigation.navs.INav
import com.vitorpamplona.amethyst.ui.navigation.topbars.TopBarWithBackButton
import com.vitorpamplona.amethyst.ui.note.timeAgoNoDot
import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
import com.vitorpamplona.amethyst.ui.stringRes
import com.vitorpamplona.amethyst.ui.theme.ThemeComparisonColumn
@@ -292,28 +294,9 @@ private fun SyncProgressCard(
elevation = CardDefaults.cardElevation(defaultElevation = 1.dp),
) {
Column(modifier = Modifier.padding(16.dp)) {
Text(
text = stringRes(R.string.event_sync_relays_progress, state.relaysCompleted, state.totalRelays),
style = MaterialTheme.typography.titleSmall,
fontWeight = FontWeight.SemiBold,
)
RelayStatement(state)
Spacer(Modifier.height(8.dp))
LinearProgressIndicator(
progress = {
if (state.totalRelays > 0) {
state.relaysCompleted.toFloat() / state.totalRelays
} else {
0f
}
},
modifier = Modifier.fillMaxWidth(),
)
Spacer(Modifier.height(8.dp))
Text(
text = stringRes(R.string.event_sync_events_sent, state.eventsSent),
style = MaterialTheme.typography.bodySmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
)
EventsReceivedStatement(state)
Spacer(Modifier.height(10.dp))
OutlinedButton(
onClick = onCancel,
@@ -329,6 +312,42 @@ private fun SyncProgressCard(
}
}
@Composable
private fun EventsReceivedStatement(state: EventSync.SyncState.Running) {
val eventsReceived by state.eventsReceived.collectAsStateWithLifecycle()
val eventsSent by state.eventsSent.collectAsStateWithLifecycle()
val eventsAccepted by state.eventsAccepted.collectAsStateWithLifecycle()
Text(
text = stringRes(R.string.event_sync_events_sent, eventsAccepted, eventsSent, eventsReceived),
style = MaterialTheme.typography.bodySmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
)
}
@Composable
private fun RelayStatement(state: EventSync.SyncState.Running) {
val relaysCompleted by state.relaysCompleted.collectAsStateWithLifecycle()
val totalRelays by state.totalRelays.collectAsStateWithLifecycle()
Text(
text = stringRes(R.string.event_sync_relays_progress, relaysCompleted, totalRelays),
style = MaterialTheme.typography.titleSmall,
fontWeight = FontWeight.SemiBold,
)
Spacer(Modifier.height(8.dp))
LinearProgressIndicator(
progress = {
if (totalRelays > 0) {
relaysCompleted / totalRelays.toFloat()
} else {
0f
}
},
modifier = Modifier.fillMaxWidth(),
)
}
@Composable
private fun DoneCard(
state: EventSync.SyncState.Done,
@@ -351,7 +370,7 @@ private fun DoneCard(
)
Spacer(Modifier.height(6.dp))
Text(
text = stringRes(R.string.event_sync_done_sent, state.totalEventsSent),
text = stringRes(R.string.event_sync_done_sent, state.totalEventsSent, state.totalEventsReceived),
style = MaterialTheme.typography.bodyMedium,
color = MaterialTheme.colorScheme.onPrimaryContainer,
)
@@ -544,7 +563,6 @@ private fun DestinationRelayRow(
@Composable
private fun ActivityLogRow(info: EventSync.LiveSyncActivity.SourceRelayInfo) {
val eventsFound by info.eventsFound.collectAsStateWithLifecycle()
val status by info.status.collectAsStateWithLifecycle()
val hasEvents = eventsFound > 0
val dotColor =
if (hasEvents) {
@@ -583,12 +601,25 @@ private fun ActivityLogRow(info: EventSync.LiveSyncActivity.SourceRelayInfo) {
modifier = Modifier.weight(0.6f),
)
if (hasEvents) {
val context = LocalContext.current
val untilPage by info.pageUntil.collectAsStateWithLifecycle()
val eventsAccepted by info.eventsAccepted.collectAsStateWithLifecycle()
untilPage?.let {
Text(
text = stringRes(R.string.event_sync_less_than_until, timeAgoNoDot(it, context)),
style = MaterialTheme.typography.bodySmall,
color = textColor,
modifier = Modifier.weight(0.3f),
maxLines = 1,
textAlign = TextAlign.End,
overflow = TextOverflow.StartEllipsis,
)
}
Text(
text = stringRes(R.string.event_sync_log_recv, formatCount(eventsFound)),
style = MaterialTheme.typography.bodySmall,
color = textColor,
modifier = Modifier.weight(0.2f),
modifier = Modifier.weight(0.3f),
maxLines = 1,
textAlign = TextAlign.End,
overflow = TextOverflow.StartEllipsis,
@@ -606,7 +637,7 @@ private fun ActivityLogRow(info: EventSync.LiveSyncActivity.SourceRelayInfo) {
maxLines = 1,
textAlign = TextAlign.End,
overflow = TextOverflow.StartEllipsis,
modifier = Modifier.weight(0.2f),
modifier = Modifier.weight(0.3f),
)
} else {
val status by info.status.collectAsStateWithLifecycle()
@@ -737,7 +768,9 @@ fun SyncProgressCardPreview() {
EventSync.SyncState.Running(
relaysCompleted = 312,
totalRelays = 1024,
eventsAccepted = 12,
eventsSent = 4821,
eventsReceived = 10000,
),
onCancel = {},
)
@@ -751,6 +784,7 @@ fun DoneCardPreview() {
DoneCard(
state =
EventSync.SyncState.Done(
totalEventsReceived = 20_000,
totalEventsSent = 18_432,
totalEventsAccepted = 14_891,
durationMs = 187_000,
@@ -791,7 +825,7 @@ fun EventScreenBodyPreview() {
fun EventScreenBody2Preview() {
ThemeComparisonRow {
EventScreenBody(
EventSync.SyncState.Running(1247, 1024, 4821),
EventSync.SyncState.Running(1047, 1224, 100, 4821, 10000),
previewActivity,
)
}
amethyst/src/main/res/values/strings.xml
+3 -2
View File
@@ -1829,11 +1829,11 @@
<string name="event_sync_start_over">Start Over</string>
<string name="event_sync_cancel">Cancel</string>
<string name="event_sync_relays_progress">Relays: %1$d / %2$d</string>
<string name="event_sync_events_sent">Events redistributed: %1$d</string>
<string name="event_sync_events_sent">Events redistributed: %1$d new out of %2$d sent and %3$d received</string>
<string name="event_sync_paused_title">Sync Paused</string>
<string name="event_sync_paused_body">Completed %1$d of %2$d relays — %3$d events redistributed so far. Tap Resume to continue.</string>
<string name="event_sync_done_title">Sync complete</string>
<string name="event_sync_done_sent">Forwarded %1$d events to destination relays.</string>
<string name="event_sync_done_sent">Forwarded %1$d events to destination relays out of %2$d received.</string>
<string name="event_sync_done_accepted">%1$d events accepted as new by destination relays.</string>
<string name="event_sync_done_duration">Completed in %1$d seconds.</string>
<string name="event_sync_error_title">Sync error</string>
@@ -1854,6 +1854,7 @@
<string name="relay_settings_lower">relay settings</string>
<string name="last_seen">Last seen %1$s ago</string>
<string name="event_sync_less_than_until">&lt;%1$s</string>
<string name="event_sync_status_connecting">Connecting</string>
<string name="event_sync_status_downloading">Downloading</string>
<string name="event_sync_status_error">Error</string>