laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-08 23:54:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(relays): merge the notification tail across accounts into one REQ

Browse Source 
Every account-level loader is a PerUserEoseManager, which opens one subscription
per user. With four accounts open that multiplies, and shared relays run out of
room: nos.lol (strfry, `max_subscriptions: 20`) answered `ERROR: too many
concurrent REQs` — 0 times before this branch, 7 with three accounts subscribed,
13 with four. The refusal arrives as a NOTICE, which carries no subscription id
and never reaches RelayReqRefusals (wired only to CLOSED), and "too many
concurrent REQs" matches none of its markers. So the relay drops those REQs while
we still believe they are live: they never EOSE and never deliver.

Notifications are `#p`-scoped, so a relay serving several accounts can be asked
about all of them in one filter naming every pubkey — same query, wider tag. The
manager moves to SingleSubEoseManager: one REQ per relay instead of one per
(account, relay), with the `since` floor taken per relay as the OLDEST of the
participating accounts' floors so widening for one can never cut another short.

Gift wraps deliberately do NOT merge. They are unsolicited and opaque to the
relay, so it cannot rate-limit them per recipient; a merged query would let one
spammed account eat the shared `limit` and starve every other account's DMs.
Each account keeps its own budget there.

A merged filter serves several accounts, so accountPubKey becomes accountPubKeys.
The subscriptions screen shows such a filter under each account it serves —
"why is this relay busy for me" has to be answerable per account — which makes
the per-account counts a breakdown of a shared filter rather than a partition of
the total. attributedFilters is that sum, and is what a card's share is drawn
against now; dividing by the wire total would read as 100% for each of two
accounts sharing one filter.

Measured, and it is only part of the answer: nos.lol carries 24-26 subscriptions
against its cap of 20, and this removes 3 of them. The rest are the other
per-account managers — account metadata alone is 7 filters in a subscription per
account, and gift wraps another. Merging metadata (it is `authors`-keyed and
merges the same way) is the next lever; this commit does not get us under the cap
on its own.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Vitor Pamplona
2026-07-31 11:09:20 -04:00
co-authored by Claude Opus 5
parent 61611bdc48
commit 24decefd4b
17 changed files with 270 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/follows/AccountFollowsLoaderSubAssembler.kt
+1 -1
View File
@@ -181,7 +181,7 @@ class AccountFollowsLoaderSubAssembler(
purpose = SubPurpose.RELAY_LISTS,
kinds = listOf(AdvertisedRelayListEvent.KIND),
authors = users.sorted(),
accountPubKey = soleAccountPubKey,
accountPubKeys = listOfNotNull(soleAccountPubKey),
),
)
} else {
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip01Notifications/AccountNotificationsEoseFromInboxRelaysManager.kt
+122 -86
View File
@@ -20,13 +20,14 @@
*/
package com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip01Notifications
import com.vitorpamplona.amethyst.commons.relayClient.eoseManagers.SingleSubEoseManager
import com.vitorpamplona.amethyst.model.User
import com.vitorpamplona.amethyst.service.relayClient.eoseManagers.PerUserEoseManager
import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.AccountQueryState
import com.vitorpamplona.amethyst.service.relays.SincePerRelayMap
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.Job
@@ -34,107 +35,142 @@ import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.sample
import kotlinx.coroutines.launch
/**
* The live notification tail, for **every** logged-in account, in one subscription.
*
* Notifications are `#p`-scoped, so a relay that serves several of the user's accounts can be asked
* about all of them in one filter naming every pubkey — the query is identical in shape, only wider.
* That is what keeps this within a relay's `max_subscriptions`: one REQ per relay instead of one per
* (account, relay). With four accounts open, the per-account form pushed strfry relays past their
* 20-subscription cap and they answered `ERROR: too many concurrent REQs` — a NOTICE, carrying no
* subscription id, so the client could not even tell which REQ had been dropped. Those filters stayed
* "live" in our books and silently never delivered.
*
* Gift wraps deliberately do **not** merge this way. They are unsolicited and their content is opaque
* to the relay, so it cannot rate-limit them per recipient; a merged query would let one spammed
* account consume the shared `limit` and starve every other account's DMs. Each account keeps its own
* budget there — see [com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip59GiftWraps.AccountGiftWrapsEoseManager].
*/
class AccountNotificationsEoseFromInboxRelaysManager(
client: INostrClient,
allKeys: () -> Set<AccountQueryState>,
) : PerUserEoseManager<AccountQueryState>(client, allKeys) {
override fun user(key: AccountQueryState) = key.account.userProfile()
) : SingleSubEoseManager<AccountQueryState>(client, allKeys) {
override fun distinct(key: AccountQueryState) = key.account.userProfile()
/**
* Downloads most notifications from the user's own inbox relays.
* But also connects to all the follows relays to check for new notifications that are not in the user's
* own inbox.
* One filter set per inbox relay, naming every account that reads from it.
*
* The `since` floor is per relay rather than per account, because the merged filter is per relay:
* the **oldest** of the participating accounts' floors wins, so widening the query for one account
* can never cut another one short.
*/
override fun updateFilter(
key: AccountQueryState,
keys: List<AccountQueryState>,
since: SincePerRelayMap?,
): List<RelayBasedFilter> {
// A cold-start floor, NOT paging — backward paging lives in
// [AccountNotificationsHistoryEoseManager]. Read only when a relay has no EOSE yet; once it
// does, the EOSE time wins and this is never consulted.
//
// `since` means *newer than*, so this floors the query at the depth the feed already reaches
// rather than asking all-time again. It stays null until the feed holds a full page — and a
// background account has no feed at all (no screen ever mounted one), which is why the key
// carries none and this reads null there.
val pagingBoundary = key.feedContentStates?.notifications?.lastNoteCreatedAtIfFilled()
val inbox =
key.account.notificationRelays.flow.value.flatMap {
// No `since` floor on the first fetch. These filters are scoped by `#p` to my own
// key and carry a relay-side `limit`, so an all-time query costs one index scan and
// returns at most `limit` events, newest first — exactly what Home does (it passes
// `since ?: boundary`, i.e. null on a cold start).
//
// This used to fall back to `oneWeekAgo()`, which silently emptied the tab for
// anyone whose last mention was older than a week: EOSE `since` is in-memory only,
// so EVERY cold start re-pinned the window to 7 days, and the paging boundary above
// could never rescue it — it only arms once the feed holds a full page, and the feed
// could not fill because the query only ever asked for a week. A fresh install of an
// established account hit the same deadlock.
val notificationSince = since?.get(it)?.time ?: pagingBoundary
filterSummaryNotificationsToPubkey(
relay = it,
pubkey = user(key).pubkeyHex,
since = notificationSince,
) +
filterNotificationsToPubkey(
relay = it,
pubkey = user(key).pubkeyHex,
since = notificationSince,
)
val accountsPerRelay = mutableMapOf<NormalizedRelayUrl, MutableList<AccountQueryState>>()
keys.forEach { key ->
key.account.notificationRelays.flow.value.forEach { relay ->
accountsPerRelay.getOrPut(relay) { mutableListOf() }.add(key)
}
}
// NIP-29 group activity (reactions/replies to my messages) is deliberately NOT requested here.
// It lives on the group's host relay and used to be one `#h` filter per relay carrying every
// joined group id — but this subscription also carries the inbox filters above, which have no
// `#h` at all, and `block/buzz` downgrades any subscription with a channel-less (or multi-
// channel) filter to "global", a class that by design never receives channel-scoped events. So
// those filters answered the stored query at EOSE and then went deaf until the next launch.
//
// Group and Buzz-DM activity now rides the per-channel subscriptions that are already scoped to
// exactly one channel — RelayGroupJoinedChatTailSubAssembler and BuzzDmJoinedChatTailSubAssembler,
// both mounted app-wide from LoggedInPage, so coverage is unchanged and delivery is live.
return inbox
return accountsPerRelay.flatMap { (relay, accounts) ->
val pubkeys = accounts.map { it.account.userProfile().pubkeyHex }
// A cold-start floor, NOT paging — backward paging lives in
// [AccountNotificationsHistoryEoseManager]. Read only when a relay has no EOSE yet; once it
// does, the EOSE time wins and this is never consulted.
//
// `since` means *newer than*, so this floors the query at the depth the feed already reaches
// rather than asking all-time again. It stays null until the feed holds a full page — and a
// background account has no feed at all (no screen ever mounted one), which is why the key
// carries none and this reads null there. Null for ANY account on the relay means no floor
// at all, since a floor derived from one account's feed would truncate the others'.
val floors = accounts.map { it.feedContentStates?.notifications?.lastNoteCreatedAtIfFilled() }
val pagingBoundary = if (floors.any { it == null }) null else floors.filterNotNull().min()
// No `since` floor on the first fetch. These filters are scoped by `#p` to my own
// keys and carry a relay-side `limit`, so an all-time query costs one index scan and
// returns at most `limit` events, newest first — exactly what Home does (it passes
// `since ?: boundary`, i.e. null on a cold start).
//
// This used to fall back to `oneWeekAgo()`, which silently emptied the tab for
// anyone whose last mention was older than a week: EOSE `since` is in-memory only,
// so EVERY cold start re-pinned the window to 7 days, and the paging boundary above
// could never rescue it — it only arms once the feed holds a full page, and the feed
// could not fill because the query only ever asked for a week. A fresh install of an
// established account hit the same deadlock.
val notificationSince = since?.get(relay)?.time ?: pagingBoundary
// NIP-29 group activity (reactions/replies to my messages) is deliberately NOT requested
// here. It lives on the group's host relay and used to be one `#h` filter per relay carrying
// every joined group id — but this subscription also carries the inbox filters below, which
// have no `#h` at all, and `block/buzz` downgrades any subscription with a channel-less (or
// multi-channel) filter to "global", a class that by design never receives channel-scoped
// events. So those filters answered the stored query at EOSE and then went deaf until the
// next launch.
//
// Group and Buzz-DM activity now rides the per-channel subscriptions that are already scoped
// to exactly one channel — RelayGroupJoinedChatTailSubAssembler and
// BuzzDmJoinedChatTailSubAssembler, both mounted app-wide from LoggedInPage, so coverage is
// unchanged and delivery is live.
filterSummaryNotificationsToPubkeys(relay = relay, pubkeys = pubkeys, since = notificationSince) +
filterNotificationsToPubkeys(relay = relay, pubkeys = pubkeys, since = notificationSince)
}
}
val userJobMap = mutableMapOf<User, List<Job>>()
/**
* Per-account watchers, rebuilt as accounts come and go.
*
* There is only one subscription now, so these cannot hang off a per-key `newSub`. They are keyed
* by user and reconciled here: an account that leaves has its jobs cancelled, and one that arrives
* gets its own. Re-entrancy is safe — the watchers call `invalidateFilters()`, which lands back
* here and finds every account already watched.
*/
private val userJobMap = mutableMapOf<User, List<Job>>()
@OptIn(FlowPreview::class)
override fun newSub(key: AccountQueryState): Subscription {
val user = user(key)
userJobMap[user]?.forEach { it.cancel() }
userJobMap[user] =
listOf(
key.account.scope.launch(Dispatchers.IO) {
key.account.notificationRelays.flow.sample(1000).collectLatest {
invalidateFilters()
}
},
// No group/Buzz-DM watchers here any more: those filters moved to the per-channel
// subscriptions, which mount and unmount with the channel itself.
) +
// Only a screen can fill a feed, so there is nothing to watch for a
// background account.
listOfNotNull(
key.feedContentStates?.let { feeds ->
key.account.scope.launch(Dispatchers.IO) {
feeds.notifications.lastNoteCreatedAtWhenFullyLoaded.sample(5000).collectLatest {
invalidateFilters()
}
}
},
)
override fun updateSubscriptions(keys: Set<AccountQueryState>) {
val wanted = keys.associateBy { it.account.userProfile() }
return super.newSub(key)
(userJobMap.keys - wanted.keys).toList().forEach { user ->
userJobMap.remove(user)?.forEach { it.cancel() }
}
wanted.forEach { (user, key) ->
if (user !in userJobMap) {
userJobMap[user] =
listOf(
key.account.scope.launch(Dispatchers.IO) {
key.account.notificationRelays.flow.sample(1000).collectLatest {
invalidateFilters()
}
},
) +
// Only a screen can fill a feed, so there is nothing to watch for a
// background account.
listOfNotNull(
key.feedContentStates?.let { feeds ->
key.account.scope.launch(Dispatchers.IO) {
feeds.notifications.lastNoteCreatedAtWhenFullyLoaded.sample(5000).collectLatest {
invalidateFilters()
}
}
},
)
}
}
super.updateSubscriptions(keys)
}
override fun endSub(
key: User,
subId: String,
) {
super.endSub(key, subId)
userJobMap[key]?.forEach { it.cancel() }
override fun destroy() {
userJobMap.values.forEach { jobs -> jobs.forEach { it.cancel() } }
userJobMap.clear()
super.destroy()
}
/** Unused here, but kept so callers reading a key's owner do not reach into the account. */
fun pubkeyOf(key: AccountQueryState): HexKey = key.account.userProfile().pubkeyHex
}
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip01Notifications/FilterNotificationsToPubkey.kt
+21 -21
View File
@@ -157,7 +157,7 @@ fun filterNotificationsHistoryToPubkey(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
kinds = AllNotificationKinds,
tags = mapOf("p" to listOf(pubkey)),
limit = limit,
@@ -186,7 +186,7 @@ fun filterGroupNotificationsHistoryToPubkey(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
kinds = GroupNotificationKinds,
tags = mapOf("p" to listOf(pubkey), "h" to groupIds),
limit = limit,
@@ -196,12 +196,12 @@ fun filterGroupNotificationsHistoryToPubkey(
)
}
fun filterSummaryNotificationsToPubkey(
fun filterSummaryNotificationsToPubkeys(
relay: NormalizedRelayUrl,
pubkey: HexKey?,
pubkeys: List<HexKey>,
since: Long?,
): List<RelayBasedFilter> {
if (pubkey.isNullOrEmpty()) return emptyList()
if (pubkeys.isEmpty()) return emptyList()
return listOf(
RelayBasedFilter(
@@ -209,9 +209,9 @@ fun filterSummaryNotificationsToPubkey(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = pubkeys,
kinds = SummaryKinds,
tags = mapOf("p" to listOf(pubkey)),
tags = mapOf("p" to pubkeys),
limit = 2000,
since = since,
),
@@ -219,12 +219,12 @@ fun filterSummaryNotificationsToPubkey(
)
}
fun filterNotificationsToPubkey(
fun filterNotificationsToPubkeys(
relay: NormalizedRelayUrl,
pubkey: HexKey?,
pubkeys: List<HexKey>,
since: Long?,
): List<RelayBasedFilter> {
if (pubkey.isNullOrEmpty()) return emptyList()
if (pubkeys.isEmpty()) return emptyList()
return listOf(
RelayBasedFilter(
@@ -232,9 +232,9 @@ fun filterNotificationsToPubkey(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = pubkeys,
kinds = NotificationsPerKeyKinds,
tags = mapOf("p" to listOf(pubkey)),
tags = mapOf("p" to pubkeys),
limit = 500,
since = since,
),
@@ -244,9 +244,9 @@ fun filterNotificationsToPubkey(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = pubkeys,
kinds = NotificationsPerKeyKinds2,
tags = mapOf("p" to listOf(pubkey)),
tags = mapOf("p" to pubkeys),
limit = 200,
since = since,
),
@@ -256,9 +256,9 @@ fun filterNotificationsToPubkey(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = pubkeys,
kinds = NotificationsPerKeyKinds3,
tags = mapOf("p" to listOf(pubkey)),
tags = mapOf("p" to pubkeys),
limit = 10,
since = since,
),
@@ -286,7 +286,7 @@ fun filterGroupNotificationsToPubkey(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
kinds = GroupNotificationKinds,
tags = mapOf("p" to listOf(pubkey), "h" to groupIds),
limit = 200,
@@ -309,7 +309,7 @@ fun filterJustTheLatestNotificationsToPubkeyFromRandomRelays(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
kinds = SummaryKinds,
tags = mapOf("p" to listOf(pubkey)),
limit = 20,
@@ -321,7 +321,7 @@ fun filterJustTheLatestNotificationsToPubkeyFromRandomRelays(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
kinds = NotificationsPerKeyKinds,
tags = mapOf("p" to listOf(pubkey)),
limit = 20,
@@ -333,7 +333,7 @@ fun filterJustTheLatestNotificationsToPubkeyFromRandomRelays(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
kinds = NotificationsPerKeyKinds2,
tags = mapOf("p" to listOf(pubkey)),
limit = 10,
@@ -345,7 +345,7 @@ fun filterJustTheLatestNotificationsToPubkeyFromRandomRelays(
filter =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
kinds = NotificationsPerKeyKinds3,
tags = mapOf("p" to listOf(pubkey)),
limit = 2,
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/watchers/FilterRepliesAndReactionsToAddresses.kt
+4 -4
View File
@@ -96,7 +96,7 @@ fun filterRepliesAndReactionsToAddresses(
filter =
ExplainedFilter(
purpose = SubPurpose.ENGAGEMENT,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = RepliesAndReactionsToAddressesKinds1,
tags = mapOf("a" to sortedList),
since = since,
@@ -109,7 +109,7 @@ fun filterRepliesAndReactionsToAddresses(
filter =
ExplainedFilter(
purpose = SubPurpose.ENGAGEMENT,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = PostsAndChatMessagesToAddresses,
tags = mapOf("a" to sortedList),
since = since,
@@ -122,7 +122,7 @@ fun filterRepliesAndReactionsToAddresses(
filter =
ExplainedFilter(
purpose = SubPurpose.ENGAGEMENT,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = DeletionKindList,
tags = mapOf("a" to sortedList),
since = since,
@@ -135,7 +135,7 @@ fun filterRepliesAndReactionsToAddresses(
filter =
ExplainedFilter(
purpose = SubPurpose.ENGAGEMENT,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = TextNoteKindList,
tags = mapOf("q" to sortedList),
since = since,
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/watchers/FilterRepliesAndReactionsToNotes.kt
+3 -3
View File
@@ -103,7 +103,7 @@ fun filterRepliesAndReactionsToNotes(
filter =
ExplainedFilter(
purpose = SubPurpose.ENGAGEMENT,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = RepliesAndReactionsKinds,
tags = mapOf("e" to sortedList),
since = since,
@@ -116,7 +116,7 @@ fun filterRepliesAndReactionsToNotes(
filter =
ExplainedFilter(
purpose = SubPurpose.ENGAGEMENT,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = RepliesAndReactionsKinds2,
tags = mapOf("e" to sortedList),
since = since,
@@ -128,7 +128,7 @@ fun filterRepliesAndReactionsToNotes(
filter =
ExplainedFilter(
purpose = SubPurpose.ENGAGEMENT,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = listOf(TextNoteEvent.KIND, CommentEvent.KIND),
tags = mapOf("q" to sortedList),
since = since,
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/user/loaders/UserOutboxFinderSubAssembler.kt
+2 -2
View File
@@ -135,7 +135,7 @@ class UserOutboxFinderSubAssembler(
kinds = relayListKinds,
authors = sortedUsers,
purpose = SubPurpose.RELAY_LISTS,
accountPubKey = soleAccountPubKey,
accountPubKeys = listOfNotNull(soleAccountPubKey),
),
)
} else {
@@ -172,7 +172,7 @@ class UserOutboxFinderSubAssembler(
authors = sortedAbandoned,
purpose = SubPurpose.RELAY_LISTS,
purposeDetail = "outbox discovery for users whose relay list we lost",
accountPubKey = soleAccountPubKey,
accountPubKeys = listOfNotNull(soleAccountPubKey),
),
)
}
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/user/watchers/FilterReportsToKey.kt
+1 -1
View File
@@ -44,7 +44,7 @@ fun filterReportsToKeysFromTrusted(
purpose = SubPurpose.MODERATION,
kinds = ReportKindList,
authors = trustedAccounts,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
tags = mapOf("p" to targets.sorted()),
since = since,
),
amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/user/watchers/FilterUserMetadataForKey.kt
+2 -2
View File
@@ -115,7 +115,7 @@ fun filterUserMetadataForKey(
filter =
ExplainedFilter(
purpose = SubPurpose.PROFILE_METADATA,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = UserMetadataForKeyKinds,
authors = firstTimers.sorted(),
),
@@ -129,7 +129,7 @@ fun filterUserMetadataForKey(
filter =
ExplainedFilter(
purpose = SubPurpose.PROFILE_METADATA,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = UserMetadataForKeyKinds,
authors = updates.sorted(),
since = minimumTime,
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/subscriptions/ActiveSubscriptionsScreen.kt
+7 -3
View File
@@ -128,7 +128,7 @@ fun ActiveSubscriptionsScreen(
item(key = "acct-${account.accountPubKey ?: "none"}") { AccountHeader(account) }
items(account.purposes, key = { "${account.accountPubKey}-${it.purpose.name}" }) { purpose ->
PurposeCard(purpose, state.totalFilters, accountViewModel, nav)
PurposeCard(purpose, state.attributedFilters, accountViewModel, nav)
}
}
}
@@ -194,7 +194,11 @@ private fun AccountHeader(account: SubscriptionAccountRow) {
@Composable
private fun PurposeCard(
row: SubscriptionPurposeRow,
totalFilters: Int,
/**
* The sum of the per-account counts, not the wire total: a merged filter serving four accounts
* contributes to four cards, so dividing by the wire total would overstate every one of them.
*/
attributedFilters: Int,
accountViewModel: AccountViewModel,
nav: INav,
) {
@@ -233,7 +237,7 @@ private fun PurposeCard(
// Share of ALL subscriptions, not of the biggest one: measuring against the biggest
// makes one bar permanently full and says nothing about how much of the app's traffic a
// purpose actually accounts for.
val share = if (totalFilters > 0) row.filterCount / totalFilters.toFloat() else 0f
val share = if (attributedFilters > 0) row.filterCount / attributedFilters.toFloat() else 0f
ShareBar(fraction = share, color = accent)
Spacer(Modifier.height(8.dp))
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/subscriptions/ActiveSubscriptionsViewModel.kt
+36 -19
View File
@@ -127,6 +127,15 @@ data class ActiveSubscriptionsState(
val totalRelays: Int = 0,
/** Filters in flight that carry no purpose — assemblers not yet migrated. Honesty, not a bug. */
val untaggedFilters: Int = 0,
/**
* The sum of the per-account counts, which is what a card's share must be drawn against.
*
* It exceeds [totalFilters] when a filter serves several accounts at once — one merged
* notifications REQ naming four pubkeys is one filter on the wire but four accounts' worth of
* explanation. Dividing a per-account count by [totalFilters] would compare the two units and
* overstate every card, which is the mistake the per-entity rows already taught once.
*/
val attributedFilters: Int = 0,
) {
/** The largest purpose, so every card can draw its share against a common scale. */
val busiestPurposeFilters: Int = accounts.flatMap { it.purposes }.maxOfOrNull { it.filterCount } ?: 0
@@ -192,27 +201,34 @@ fun aggregateSubscriptions(filtersByRelay: Map<NormalizedRelayUrl, List<Filter>>
// is what keeps them from collapsing into one nameless row per purpose.
val scopeKey = explained.scope?.let { it::class.simpleName }
val tally =
byAccount
.getOrPut(explained.accountPubKey) { mutableMapOf() }
.getOrPut(explained.purpose) { PurposeTally() }
// A merged filter serves several accounts at once — notifications are `#p`-scoped, so
// every account reading a relay is asked for in one filter naming all of them. It shows
// under each of those accounts, because "why is this relay busy for me" has to be
// answerable per account. That makes the per-account counts a breakdown of a shared
// filter rather than a partition of the total, which is what [attributedFilters] exists
// to keep straight.
val accounts: List<HexKey?> = explained.accountPubKeys?.takeIf { it.isNotEmpty() } ?: listOf(null)
accounts.forEach { account ->
val tally =
byAccount
.getOrPut(account) { mutableMapOf() }
.getOrPut(explained.purpose) { PurposeTally() }
// The filter counts ONCE, here — before it is fanned out below. This is the
// number that shares a unit with `total`, so a card's share of the whole is a
// comparison of like with like.
tally.filters++
tally.relays.add(relay)
// Once per account this filter serves, never once per entity it names.
tally.filters++
tally.relays.add(relay)
// A batched filter serves several entities at once — relay-group state is one #d
// filter per host relay carrying every joined group on it — so it contributes a
// row to each of them rather than collapsing to "All". These rows are a
// breakdown, never a total: see [SubscriptionEntityRow.namedInFilters].
val entities: List<HexKey?> = explained.entityIds?.takeIf { it.isNotEmpty() } ?: listOf(null)
entities.forEach { entityId ->
val key = EntityKey(entityId, scopeKey)
tally.entities.getOrPut(key) { mutableListOf() }.add(relay)
detailOf[explained.purpose to key] = explained.purposeDetail
explained.scope?.let { scopeOf.getOrPut(explained.purpose to key) { it } }
// A batched filter serves several entities at once — relay-group state is one #d
// filter per host relay carrying every joined group on it — so it contributes a
// row to each of them rather than collapsing to "All". These rows are a
// breakdown, never a total: see [SubscriptionEntityRow.namedInFilters].
val entities: List<HexKey?> = explained.entityIds?.takeIf { it.isNotEmpty() } ?: listOf(null)
entities.forEach { entityId ->
val key = EntityKey(entityId, scopeKey)
tally.entities.getOrPut(key) { mutableListOf() }.add(relay)
detailOf[explained.purpose to key] = explained.purposeDetail
explained.scope?.let { scopeOf.getOrPut(explained.purpose to key) { it } }
}
}
}
}
@@ -258,5 +274,6 @@ fun aggregateSubscriptions(filtersByRelay: Map<NormalizedRelayUrl, List<Filter>>
totalFilters = total,
totalRelays = allRelays.size,
untaggedFilters = untagged,
attributedFilters = accounts.sumOf { it.filterCount },
)
}
amethyst/src/test/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/subscriptions/AggregateSubscriptionsTest.kt
+44 -1
View File
@@ -50,7 +50,7 @@ class AggregateSubscriptionsTest {
entityIds = chats,
kinds = listOf(kind),
tags = mapOf("e" to chats),
accountPubKey = account,
accountPubKeys = listOfNotNull(account),
)
/** Six relays, each carrying four batched filters that each name all six chats. */
@@ -99,6 +99,49 @@ class AggregateSubscriptionsTest {
)
}
/** The merged notifications filter: one REQ per relay naming every account that reads it. */
private fun mergedNotifications(vararg accounts: String) =
ExplainedFilter(
purpose = SubPurpose.NOTIFICATIONS,
accountPubKeys = accounts.toList(),
kinds = listOf(1),
tags = mapOf("p" to accounts.toList()),
)
@Test
fun `a merged filter explains itself to every account it serves`() {
val a = "aa".repeat(32)
val b = "bb".repeat(32)
val c = "cc".repeat(32)
val state = aggregateSubscriptions(mapOf(relay(1) to listOf(mergedNotifications(a, b, c))))
// One filter on the wire...
assertEquals(1, state.totalFilters)
// ...but all three accounts can see why their relay is busy.
assertEquals(3, state.accounts.size)
state.accounts.forEach { assertEquals(1, it.filterCount) }
}
@Test
fun `a card's share is drawn against the attributed total, not the wire total`() {
val a = "aa".repeat(32)
val b = "bb".repeat(32)
val state = aggregateSubscriptions(mapOf(relay(1) to listOf(mergedNotifications(a, b))))
// Dividing the per-account count by totalFilters would read as 100% for each of two
// accounts. attributedFilters is the unit those counts actually belong to.
assertEquals(1, state.totalFilters)
assertEquals(2, state.attributedFilters)
assertEquals(state.attributedFilters, state.accounts.sumOf { it.filterCount })
}
@Test
fun `unmerged filters keep attributed and wire totals identical`() {
val state = aggregateSubscriptions(sixChatsOnSixRelays())
assertEquals(state.totalFilters, state.attributedFilters)
}
@Test
fun `untagged filters are reported but never attributed to a purpose`() {
val state =
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlanner.kt
+3 -3
View File
@@ -183,7 +183,7 @@ object ConcordSubscriptionPlanner {
purpose = SubPurpose.COMMUNITY_CHATS,
purposeDetail = "concord community planes",
entityIds = listOf(entry.id),
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = listOf(ConcordStreamEnvelope.KIND_WRAP),
authors = authors.toList(),
// -1 so the last-read message (created_at == lastRead) is itself returned:
@@ -196,7 +196,7 @@ object ConcordSubscriptionPlanner {
purpose = SubPurpose.COMMUNITY_CHATS,
purposeDetail = "concord community planes",
entityIds = listOf(entry.id),
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = listOf(ConcordStreamEnvelope.KIND_WRAP),
authors = authors.toList(),
limit = previewLimit,
@@ -288,7 +288,7 @@ object ConcordSubscriptionPlanner {
purpose = SubPurpose.COMMUNITY_CHATS,
purposeDetail = "concord live planes",
entityIds = communitiesByRelay[relay]?.sorted(),
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
// Stored plane wraps (1059) plus ephemeral ones (21059) — the latter carry the
// live-only typing heartbeats a relay broadcasts but never stores.
kinds = listOf(ConcordStreamEnvelope.KIND_WRAP, ConcordStreamEnvelope.KIND_WRAP_EPHEMERAL),
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/CashuWalletFilterAssembler.kt
+2 -2
View File
@@ -138,7 +138,7 @@ private class CashuWalletSubAssembler(
MintRecommendationEvent.KIND,
),
authors = listOf(pubkey),
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
)
val inboundNutzapsFilter =
@@ -146,7 +146,7 @@ private class CashuWalletSubAssembler(
purpose = SubPurpose.NUTZAP_INBOX,
kinds = listOf(NutzapEvent.KIND),
tags = mapOf("p" to listOf(pubkey)),
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
)
// Own NIP-60 events are read from the user's outbox; inbound nutzaps
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/ContactCardFilters.kt
+2 -2
View File
@@ -48,7 +48,7 @@ fun filterContactCardsToTargetKeysFromTrustedAccountsInTheRelay(
filter =
ExplainedFilter(
purpose = SubPurpose.PROFILE_METADATA,
accountPubKey = accountPubKey,
accountPubKeys = listOfNotNull(accountPubKey),
kinds = ContactCardKindList,
authors = trustedAccounts,
// kind:30382 addresses the target user in the d-tag
@@ -75,7 +75,7 @@ fun filterContactCardsByAuthorInTheRelay(
ExplainedFilter(
purpose = SubPurpose.PROFILE_METADATA,
// This variant fetches an account's OWN contact card, so the author is the owner.
accountPubKey = author,
accountPubKeys = listOfNotNull(author),
kinds = ContactCardKindList,
authors = listOf(author),
limit = limit,
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/MetadataFilterAssembler.kt
+1 -1
View File
@@ -85,7 +85,7 @@ class MetadataFilterAssembler(
authors = pubkeyList,
limit = pubkeyList.size,
purpose = SubPurpose.PROFILE_METADATA,
accountPubKey = soleAccountPubKey,
accountPubKeys = listOfNotNull(soleAccountPubKey),
)
// Apply since times per relay
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/subscriptions/ExplainedFilter.kt
+17 -12
View File
@@ -82,15 +82,20 @@ class ExplainedFilter(
*/
val entityIds: List<HexKey>? = null,
/**
* Which logged-in account asked for this. Several accounts are commonly active at once and they
* Which logged-in accounts asked for this. Several accounts are commonly active at once and they
* do not share relay sets, so "why is this relay connected" is only answerable per account —
* without it, one account's communities look like another's.
*
* A pubkey rather than an account reference: filters outlive the objects that created them and
* are held by the relay pool for the session, so holding an `Account` here would keep a logged-out
* A **list**, because a filter can legitimately serve several accounts at once. Notifications are
* `#p`-scoped, so every account reading the same relay can be asked for in one filter naming all
* of them — which is what keeps a relay under its `max_subscriptions` cap when the app holds four
* accounts open. Filters that serve one account carry a single-element list.
*
* Pubkeys rather than account references: filters outlive the objects that created them and are
* held by the relay pool for the session, so holding an `Account` here would keep a logged-out
* account alive.
*/
val accountPubKey: HexKey? = null,
val accountPubKeys: List<HexKey>? = null,
/**
* The top-nav selection that produced this filter — Global, the user's follows, a hashtag, a
* geohash, a community.
@@ -120,7 +125,7 @@ class ExplainedFilter(
until: Long?,
limit: Int?,
search: String?,
) = ExplainedFilter(ids, authors, kinds, tags, tagsAll, since, until, limit, search, purpose, purposeDetail, entityIds, accountPubKey, scope)
) = ExplainedFilter(ids, authors, kinds, tags, tagsAll, since, until, limit, search, purpose, purposeDetail, entityIds, accountPubKeys, scope)
companion object {
/** Tags [filter] with a [purpose], preserving every protocol field. */
@@ -129,7 +134,7 @@ class ExplainedFilter(
purpose: SubPurpose,
detail: String? = null,
entityIds: List<HexKey>? = null,
accountPubKey: HexKey? = null,
accountPubKeys: List<HexKey>? = null,
scope: IFeedTopNavPerRelayFilter? = null,
) = ExplainedFilter(
filter.ids,
@@ -144,7 +149,7 @@ class ExplainedFilter(
purpose,
detail,
entityIds,
accountPubKey,
accountPubKeys,
scope,
)
}
@@ -168,9 +173,9 @@ fun Collection<Filter>.purposeEntities(): Set<PurposeEntity> =
val explained = filter as? ExplainedFilter ?: return@flatMapTo emptyList()
val ids = explained.entityIds
if (ids.isNullOrEmpty()) {
listOf(PurposeEntity(explained.purpose, null, explained.accountPubKey, explained.purposeDetail))
listOf(PurposeEntity(explained.purpose, null, explained.accountPubKeys?.firstOrNull(), explained.purposeDetail))
} else {
ids.map { PurposeEntity(explained.purpose, it, explained.accountPubKey, explained.purposeDetail) }
ids.map { PurposeEntity(explained.purpose, it, explained.accountPubKeys?.firstOrNull(), explained.purposeDetail) }
}
}
@@ -192,10 +197,10 @@ data class PurposeEntity(
fun List<RelayBasedFilter>.attributedTo(accountPubKey: HexKey): List<RelayBasedFilter> =
map { relayFilter ->
val filter = relayFilter.filter
if (filter is ExplainedFilter && filter.accountPubKey == null) {
if (filter is ExplainedFilter && filter.accountPubKeys.isNullOrEmpty()) {
RelayBasedFilter(
relay = relayFilter.relay,
filter = ExplainedFilter.of(filter, filter.purpose, filter.purposeDetail, filter.entityIds, accountPubKey, filter.scope),
filter = ExplainedFilter.of(filter, filter.purpose, filter.purposeDetail, filter.entityIds, listOf(accountPubKey), filter.scope),
)
} else {
relayFilter
@@ -219,7 +224,7 @@ fun List<RelayBasedFilter>.scopedTo(feedSettings: IFeedTopNavPerRelayFilterSet):
if (filter is ExplainedFilter && scope != null) {
RelayBasedFilter(
relay = relayFilter.relay,
filter = ExplainedFilter.of(filter, filter.purpose, filter.purposeDetail, filter.entityIds, filter.accountPubKey, scope),
filter = ExplainedFilter.of(filter, filter.purpose, filter.purposeDetail, filter.entityIds, filter.accountPubKeys, scope),
)
} else {
relayFilter
commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/subscriptions/ExplainedFilterTest.kt
+2 -2
View File
@@ -58,7 +58,7 @@ class ExplainedFilterTest {
purpose = SubPurpose.NOTIFICATIONS,
purposeDetail = "inbox relays for the active account",
entityIds = listOf("cafe0000000000000000000000000000000000000000000000000000000000ff"),
accountPubKey = pubkey,
accountPubKeys = listOfNotNull(pubkey),
scope = HashtagTopNavPerRelayFilter(setOf("askednostr")),
)
@@ -115,7 +115,7 @@ class ExplainedFilterTest {
assertEquals("inbox relays for the active account", (advanced as ExplainedFilter).purposeDetail)
// entityIds/accountPubKey/scope ride the same path and would vanish just as silently
assertEquals(listOf("cafe0000000000000000000000000000000000000000000000000000000000ff"), advanced.entityIds)
assertEquals(pubkey, advanced.accountPubKey)
assertEquals(listOf(pubkey), advanced.accountPubKeys)
assertEquals(setOf("askednostr"), (advanced.scope as? HashtagTopNavPerRelayFilter)?.hashtags)
assertEquals(1_785_379_272L, advanced.since)
// and the protocol fields came along untouched