mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-10-06 03:38:23 +00:00
Merge pull request #3879 from davotoula/feat/relay-usage-diagnostics
feat(resourceusage): measure why relay data is spent, not just how much
This commit is contained in:
Vitor Pamplona
@@ -723,6 +723,7 @@ class AppModules(
|
||||
torManager.status,
|
||||
client,
|
||||
applicationIOScope,
|
||||
onTrigger = { cause -> resourceUsage.add(UsageKeys.relayTrigger(cause), 1) },
|
||||
)
|
||||
|
||||
// Verifies and inserts in the cache from all relays, all subscriptions
|
||||
|
||||
+29
@@ -23,6 +23,7 @@ package com.vitorpamplona.amethyst.service.relayClient
|
||||
import com.vitorpamplona.amethyst.commons.tor.TorRelaySettings
|
||||
import com.vitorpamplona.amethyst.model.torState.TorRelayEvaluation
|
||||
import com.vitorpamplona.amethyst.service.connectivity.ConnectivityStatus
|
||||
import com.vitorpamplona.amethyst.service.resourceusage.UsageKeys
|
||||
import com.vitorpamplona.amethyst.ui.tor.TorServiceStatus
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
@@ -49,6 +50,14 @@ class RelayProxyClientConnector(
|
||||
val torStatus: StateFlow<TorServiceStatus>,
|
||||
val client: INostrClient,
|
||||
val scope: CoroutineScope,
|
||||
/**
|
||||
* Called with the cause every time this connector *decides* to reconnect, so the
|
||||
* usage ledger can attribute relay churn without this class knowing where the
|
||||
* counters go. Causes are the `UsageKeys.TRIGGER_*` constants — see
|
||||
* [UsageKeys.relayTrigger] for why these are an upper bound rather than a count
|
||||
* of reconnects actually performed.
|
||||
*/
|
||||
val onTrigger: (String) -> Unit = {},
|
||||
) {
|
||||
data class RelayServiceInfra(
|
||||
val evaluator: TorRelayEvaluation,
|
||||
@@ -138,6 +147,9 @@ class RelayProxyClientConnector(
|
||||
infra.connectivity is ConnectivityStatus.Off -> {
|
||||
Log.d("ManageRelayServices") { "Connectivity Off: Pausing Relay Services ${infra.connectivity}" }
|
||||
if (client.isActive()) {
|
||||
// Counted inside the guard: the upstream combine() re-emits Off
|
||||
// repeatedly and only this branch does any work.
|
||||
onTrigger(UsageKeys.TRIGGER_OFF)
|
||||
client.disconnect()
|
||||
}
|
||||
if (infra.torStatus is TorServiceStatus.Active) {
|
||||
@@ -156,6 +168,7 @@ class RelayProxyClientConnector(
|
||||
}
|
||||
|
||||
// only calls this if the client is not active. Otherwise goes to the else below
|
||||
onTrigger(UsageKeys.TRIGGER_COLD_START)
|
||||
client.connect()
|
||||
lastNetworkId = networkId
|
||||
lastTorSettings = torSettings
|
||||
@@ -211,10 +224,26 @@ class RelayProxyClientConnector(
|
||||
Log.d("ManageRelayServices") {
|
||||
"Network identity changed ($previousNetworkId -> $networkId), rebuilding every relay connection"
|
||||
}
|
||||
// The expensive branch, and the one the churn investigation is
|
||||
// aimed at: a full teardown re-dials the whole pool and replays
|
||||
// every REQ.
|
||||
//
|
||||
// Only this cause is counted here, so a wifi<->cellular handoff —
|
||||
// which mints a new network handle AND rebuilds the OkHttp clients
|
||||
// off the metered bit — is booked as netid alone. relay.trigger.transport
|
||||
// therefore undercounts exactly the case one would most want it for;
|
||||
// read it as "transport changed WITHOUT the handle changing".
|
||||
onTrigger(UsageKeys.TRIGGER_NETID)
|
||||
// Full teardown: disconnect() drops the dead sockets AND clears each
|
||||
// relay's backoff, so the new network starts from a clean slate.
|
||||
client.reconnect(onlyIfChanged = false, ignoreRetryDelays = true)
|
||||
} else {
|
||||
// Non-exclusive: count each independently so the report shows
|
||||
// which combination fired.
|
||||
if (transportChanged) onTrigger(UsageKeys.TRIGGER_TRANSPORT)
|
||||
if (torPolicyChanged) onTrigger(UsageKeys.TRIGGER_TOR_POLICY)
|
||||
if (classificationChanged) onTrigger(UsageKeys.TRIGGER_CLASSIFICATION)
|
||||
|
||||
val freshStart = transportChanged || torPolicyChanged
|
||||
if (freshStart) {
|
||||
// The failures behind the current backoffs were measured against a
|
||||
|
||||
+293
-4
@@ -20,10 +20,22 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.service.resourceusage
|
||||
|
||||
import android.os.SystemClock
|
||||
import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.purposeOrNull
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.utils.Log
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* Counts relay websocket traffic into the usage ledger. Frame sizes are
|
||||
@@ -31,20 +43,145 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
|
||||
* (relay JSON is ASCII-dominant), consistent with how RelayStats counts.
|
||||
* Excludes WS framing/compression; good enough for "which subsystem is
|
||||
* eating my data plan" comparisons.
|
||||
*
|
||||
* Beyond the byte totals this also carries the relay-churn diagnostics: a
|
||||
* per-verb split of those same bytes, a connection-lifetime histogram, and
|
||||
* per-relay failure/short-session counts. See
|
||||
* plans/2026-07-29-relay-churn-diagnostics.md for what each answers and how to
|
||||
* read them together.
|
||||
*
|
||||
* The verb split takes its name straight from the wire label the command already
|
||||
* knows (`Command.label()` / `Message.label()`), so `Σ verb == Σ msg` holds by
|
||||
* construction and a new subtype needs no change here.
|
||||
*/
|
||||
class RelayUsageListener(
|
||||
private val accountant: ResourceUsageAccountant,
|
||||
private val isMobile: () -> Boolean,
|
||||
private val isForeground: () -> Boolean,
|
||||
private val nowMs: () -> Long = { SystemClock.elapsedRealtime() },
|
||||
) : RelayConnectionListener {
|
||||
/**
|
||||
* Relay -> when its current session became ready. Touched from the per-relay
|
||||
* OkHttp dispatcher threads, hence concurrent. Keyed by [NormalizedRelayUrl] to
|
||||
* match the other per-relay caches (`RelayStats`, `RelayLimitsTracker`).
|
||||
*
|
||||
* Entries are consumed on disconnect. Three ways a session escapes the map
|
||||
* unrecorded, all counted rather than prevented — see [UsageKeys.RELAY_LIFE_OVERWRITE],
|
||||
* [UsageKeys.RELAY_LIFE_ORPHAN], and [UsageKeys.relayConnects] for process death.
|
||||
*/
|
||||
private val connectedSince = ConcurrentHashMap<NormalizedRelayUrl, Long>()
|
||||
|
||||
/**
|
||||
* Relay -> subscription ids currently open on this connection, so a REQ that
|
||||
* replaces an in-flight subscription can be told apart from one that opens a new
|
||||
* one. Cleared on disconnect, because the relay forgets them too — every REQ
|
||||
* after a reconnect is legitimately new.
|
||||
*
|
||||
* Bounded by the live subscription count per relay (tens), not by session length.
|
||||
*/
|
||||
private val openSubs = ConcurrentHashMap<NormalizedRelayUrl, MutableSet<String>>()
|
||||
|
||||
/**
|
||||
* Subscription id -> the purpose that opened it, for attributing inbound frames:
|
||||
* an EVENT names only its subscription, never why the client asked for it.
|
||||
*
|
||||
* Deliberately **not** [openSubs]. Sharing one map conflated two different
|
||||
* lifetimes and lost 41 % of the download to `unattributed` in the 2026-08-02
|
||||
* reading: a CLOSE removed the id, and a disconnect dropped the whole relay's
|
||||
* map, while frames already in flight were still arriving. "Is this subscription
|
||||
* open" and "what did this subscription belong to" answer different questions and
|
||||
* expire at different times — the second stays true after the first turns false.
|
||||
*
|
||||
* Keyed by subscription id alone, without the relay. The same id is used across
|
||||
* relays for one logical subscription, so the purpose is a property of the id;
|
||||
* this also means a frame arriving after a reconnect still attributes.
|
||||
*
|
||||
* Bounded by [MAX_TRACKED_SUBS] with wholesale eviction rather than an LRU: this
|
||||
* is a diagnostic on a hot path, ids are recycled steadily, and a rare reset that
|
||||
* sends a few frames to `unattributed` is cheaper than per-frame bookkeeping.
|
||||
* `unattributed` staying small is what says the bound is generous enough.
|
||||
*/
|
||||
private val subPurpose = ConcurrentHashMap<String, String>()
|
||||
|
||||
/**
|
||||
* Event ids delivered recently, as the first 64 bits of the id.
|
||||
*
|
||||
* Held as a Long rather than the 64-char hex, which saves the id strings
|
||||
* themselves — but a boxed `Long` plus its map node still costs ~56 bytes an
|
||||
* entry, so a full window is ~3 MB, not the few hundred KB the raw payload
|
||||
* suggests. Worth knowing before raising [MAX_TRACKED_EVENTS] on a 512 MB-class
|
||||
* device; an unboxed open-addressed `LongArray` would be ~400 KB if it ever needs
|
||||
* to grow. 64 bits makes a collision between distinct ids negligible where a
|
||||
* 32-bit hash would not be.
|
||||
*
|
||||
* The window only needs to span the fan-out, not the session: the same event
|
||||
* arrives from every relay carrying it within seconds, so near-term memory
|
||||
* catches the duplication this measures. Cleared wholesale at [MAX_TRACKED_EVENTS]
|
||||
* for the same reason [subPurpose] is — the alternative is per-frame LRU
|
||||
* bookkeeping on the hottest path in the app. A clear undercounts duplicates that
|
||||
* straddle it, so the ratio is a floor.
|
||||
*/
|
||||
private val recentEventIds = ConcurrentHashMap.newKeySet<Long>()
|
||||
|
||||
/**
|
||||
* Relay -> when this dial was decided, so the pre-request cost can be separated
|
||||
* from the handshake. Consumed by whichever of `onConnected`/`onCannotConnect`
|
||||
* ends the dial, so an entry never outlives the attempt that made it.
|
||||
*/
|
||||
private val dialStartedAt = ConcurrentHashMap<NormalizedRelayUrl, Long>()
|
||||
|
||||
/** Notice texts already logged, so one wording costs one line however often it arrives. */
|
||||
private val loggedNotices = ConcurrentHashMap.newKeySet<String>()
|
||||
|
||||
override fun onSent(
|
||||
relay: IRelayClient,
|
||||
cmdStr: String,
|
||||
cmd: Command,
|
||||
success: Boolean,
|
||||
) {
|
||||
if (success) {
|
||||
accountant.add(UsageKeys.relayMsg(isMobile(), isForeground(), received = false), cmdStr.length.toLong())
|
||||
if (!success) return
|
||||
|
||||
val bytes = cmdStr.length.toLong()
|
||||
val mobile = isMobile()
|
||||
val fg = isForeground()
|
||||
accountant.add(UsageKeys.relayMsg(mobile, fg, received = false), bytes)
|
||||
accountant.add(UsageKeys.relayVerb(cmd.label(), received = false, mobile, fg), bytes)
|
||||
|
||||
when (cmd) {
|
||||
is ReqCmd -> {
|
||||
accountant.add(UsageKeys.relaySubsSent(mobile, fg), 1)
|
||||
|
||||
val purpose = purposeOf(cmd)
|
||||
accountant.add(UsageKeys.relayPurposeSent(purpose), 1)
|
||||
accountant.add(UsageKeys.relayPurposeBytes(purpose), bytes)
|
||||
|
||||
if (subPurpose.size >= MAX_TRACKED_SUBS) subPurpose.clear()
|
||||
subPurpose[cmd.subId] = purpose
|
||||
|
||||
// Already open on this connection, so this REQ replaces a live
|
||||
// subscription rather than starting one.
|
||||
// computeIfAbsent, not getOrPut: the latter is get-then-put and two
|
||||
// threads racing the first REQ after a (re)connect would each build a
|
||||
// set, the losing put's subId vanishing with its orphaned set.
|
||||
// `sendIfConnected` deliberately calls listeners outside PoolRequests'
|
||||
// stripe lock, so same-relay concurrency here is by design.
|
||||
val known = openSubs.computeIfAbsent(relay.url) { ConcurrentHashMap.newKeySet() }
|
||||
if (!known.add(cmd.subId)) {
|
||||
accountant.add(UsageKeys.relaySubsResent(mobile, fg), 1)
|
||||
}
|
||||
// Within the window after this relay's connect, so almost certainly
|
||||
// part of syncState's replay rather than a user action. A time
|
||||
// window because nothing on this side marks a frame as belonging to
|
||||
// it; see UsageKeys.relaySubsReplay.
|
||||
val since = connectedSince[relay.url]
|
||||
if (since != null && nowMs() - since <= UsageKeys.REPLAY_WINDOW_MS) {
|
||||
accountant.add(UsageKeys.relaySubsReplay(mobile, fg), 1)
|
||||
}
|
||||
}
|
||||
is CloseCmd -> {
|
||||
accountant.add(UsageKeys.relaySubsClosed(mobile, fg), 1)
|
||||
openSubs[relay.url]?.remove(cmd.subId)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -53,7 +190,63 @@ class RelayUsageListener(
|
||||
msgStr: String,
|
||||
msg: Message,
|
||||
) {
|
||||
accountant.add(UsageKeys.relayMsg(isMobile(), isForeground(), received = true), msgStr.length.toLong())
|
||||
val bytes = msgStr.length.toLong()
|
||||
val mobile = isMobile()
|
||||
val fg = isForeground()
|
||||
accountant.add(UsageKeys.relayMsg(mobile, fg, received = true), bytes)
|
||||
accountant.add(UsageKeys.relayVerb(msg.label(), received = true, mobile, fg), bytes)
|
||||
|
||||
// Attribute the inbound side to whoever asked for it. Only frames that name a
|
||||
// subscription can be attributed; NOTICE and OK are relay-wide and are left out
|
||||
// rather than guessed at, which is why this does not reconcile to msg.rx.
|
||||
// Resolved once: the duplicate check below needs the same answer, and an
|
||||
// EVENT names only its subscription, never why the client asked for it.
|
||||
val purpose = subIdOf(msg)?.let { subPurpose[it] ?: UsageKeys.PURPOSE_UNATTRIBUTED }
|
||||
if (purpose != null) {
|
||||
accountant.add(UsageKeys.relayPurposeDown(purpose), bytes)
|
||||
accountant.add(UsageKeys.relayPurposeDownCount(purpose), 1)
|
||||
}
|
||||
|
||||
if (msg is EventMessage) {
|
||||
accountant.add(UsageKeys.relayEventsSeen(mobile, fg), 1)
|
||||
idPrefix(msg.event.id)?.let { key ->
|
||||
if (recentEventIds.size >= MAX_TRACKED_EVENTS) recentEventIds.clear()
|
||||
if (!recentEventIds.add(key)) {
|
||||
accountant.add(UsageKeys.relayEventsDup(mobile, fg), 1)
|
||||
accountant.add(UsageKeys.relayEventsDupBytes(mobile, fg), bytes)
|
||||
if (purpose != null) accountant.add(UsageKeys.relayPurposeDupBytes(purpose), bytes)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A refused subscription arrives here and nowhere else: the NOTICE carries no
|
||||
// subscription id, so RelayReqRefusals (wired to CLOSED) never sees it.
|
||||
if (msg is NoticeMessage) {
|
||||
val reason = UsageKeys.noticeReason(msg.message)
|
||||
accountant.add(UsageKeys.relayNotice(reason), 1)
|
||||
if (reason == UsageKeys.NOTICE_UNCLASSIFIED) {
|
||||
// The counter alone cannot say whether an absent `toomanysubs` means no
|
||||
// refusals or an allowlist that misses how this relay words them.
|
||||
//
|
||||
// INFO, not DEBUG: a debug build defaults to LogLevel.INFO
|
||||
// (Amethyst.DEFAULT_LOG_LEVEL, with VERBOSE_LOGS off), so a DEBUG line
|
||||
// here is dropped before it reaches the sink and this said nothing at
|
||||
// all. Demote it once the allowlist stops needing evidence.
|
||||
//
|
||||
// One line per distinct wording rather than per frame: the ledger
|
||||
// already has the count, what is missing is the variety. Truncated and
|
||||
// capped because the text is server-controlled.
|
||||
if (loggedNotices.size < MAX_DISTINCT_NOTICES && loggedNotices.add(msg.message)) {
|
||||
Log.i(TAG) { "Unclassified NOTICE from ${relay.url.url}: ${msg.message.take(MAX_NOTICE_LOG)}" }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Dial attempts. Unlike [onCannotConnect] this really is one per dial. */
|
||||
override fun onConnecting(relay: IRelayClient) {
|
||||
accountant.add(UsageKeys.relayDials(isMobile(), isForeground()), 1)
|
||||
dialStartedAt[relay.url] = nowMs()
|
||||
}
|
||||
|
||||
// Every completed (re)connection paid a TCP+TLS handshake; high daily
|
||||
@@ -64,13 +257,109 @@ class RelayUsageListener(
|
||||
pingMillis: Int,
|
||||
compressed: Boolean,
|
||||
) {
|
||||
accountant.add(UsageKeys.relayConnects(isMobile(), isForeground()), 1)
|
||||
val mobile = isMobile()
|
||||
val fg = isForeground()
|
||||
// Doubles as the lifetime histogram's denominator — one session begins here.
|
||||
accountant.add(UsageKeys.relayConnects(mobile, fg), 1)
|
||||
// Consumed unconditionally: the gap needs a handshake to subtract, but the
|
||||
// stamp has to go either way or a dial that cannot be timed leaks its entry.
|
||||
val dialedAt = dialStartedAt.remove(relay.url)
|
||||
// pingMillis is the transport's own handshake timing; <= 0 means it could
|
||||
// not measure it, and a fabricated 0 would be worse than no record.
|
||||
if (pingMillis > 0) {
|
||||
accountant.add(UsageKeys.relayHandshake(pingMillis.toLong(), mobile, fg), 1)
|
||||
if (dialedAt != null) {
|
||||
val gap = nowMs() - dialedAt - pingMillis
|
||||
if (gap >= 0) accountant.add(UsageKeys.relayDialGap(gap, mobile, fg), 1)
|
||||
}
|
||||
}
|
||||
|
||||
if (connectedSince.put(relay.url, nowMs()) != null) {
|
||||
accountant.add(UsageKeys.RELAY_LIFE_OVERWRITE, 1)
|
||||
}
|
||||
}
|
||||
|
||||
override fun onDisconnected(relay: IRelayClient) {
|
||||
val mobile = isMobile()
|
||||
val fg = isForeground()
|
||||
accountant.add(UsageKeys.relayDisconnects(mobile, fg), 1)
|
||||
|
||||
openSubs.remove(relay.url)
|
||||
val startedAt = connectedSince.remove(relay.url)
|
||||
if (startedAt == null) {
|
||||
// A dial that never became ready, or a second disconnect for one session.
|
||||
accountant.add(UsageKeys.RELAY_LIFE_ORPHAN, 1)
|
||||
return
|
||||
}
|
||||
|
||||
val elapsed = (nowMs() - startedAt).coerceAtLeast(0)
|
||||
accountant.add(UsageKeys.relayLife(elapsed, mobile, fg), 1)
|
||||
}
|
||||
|
||||
/**
|
||||
* The [SubPurpose] behind a REQ, from the first filter that declares one.
|
||||
*
|
||||
* One subscription id carries one purpose in practice, so the first is the
|
||||
* subscription's. A REQ whose filters are plain [com.vitorpamplona.quartz.nip01Core.relay.filters.Filter]s
|
||||
* predates #3832's tagging and is counted separately rather than guessed at.
|
||||
*/
|
||||
private fun purposeOf(cmd: ReqCmd): String =
|
||||
cmd.filters
|
||||
.firstNotNullOfOrNull { it.purposeOrNull() }
|
||||
?.let { UsageKeys.purposeKeyPart(it) }
|
||||
?: UsageKeys.PURPOSE_UNEXPLAINED
|
||||
|
||||
/**
|
||||
* The first 64 bits of an event id, or null if it is not a well-formed id.
|
||||
*
|
||||
* Parsed in place rather than `substring(0, 16).toULongOrNull(16)`: this runs on
|
||||
* every inbound EVENT frame, and the substring would be a String plus its backing
|
||||
* array per event, thrown away immediately.
|
||||
*/
|
||||
private fun idPrefix(id: String): Long? {
|
||||
if (id.length < 16) return null
|
||||
var acc = 0L
|
||||
for (i in 0 until 16) {
|
||||
val digit = Character.digit(id[i], 16)
|
||||
if (digit < 0) return null
|
||||
acc = (acc shl 4) or digit.toLong()
|
||||
}
|
||||
return acc
|
||||
}
|
||||
|
||||
/** The subscription a frame belongs to, when it names one. */
|
||||
private fun subIdOf(msg: Message): String? =
|
||||
when (msg) {
|
||||
is EventMessage -> msg.subId
|
||||
is EoseMessage -> msg.subId
|
||||
is ClosedMessage -> msg.subId
|
||||
is CountMessage -> msg.queryId
|
||||
else -> null
|
||||
}
|
||||
|
||||
override fun onCannotConnect(
|
||||
relay: IRelayClient,
|
||||
errorMessage: String,
|
||||
) {
|
||||
accountant.add(UsageKeys.relayConnectFails(isMobile(), isForeground()), 1)
|
||||
// A dial that ends here never reaches onConnected, so nothing else would
|
||||
// ever consume its start stamp.
|
||||
dialStartedAt.remove(relay.url)
|
||||
}
|
||||
|
||||
companion object {
|
||||
private const val TAG = "RelayUsage"
|
||||
|
||||
/** NOTICE text is server-controlled; cap what reaches the log. */
|
||||
private const val MAX_NOTICE_LOG = 200
|
||||
|
||||
/** Ceiling on distinct wordings held in memory; relay prose is unbounded. */
|
||||
private const val MAX_DISTINCT_NOTICES = 200
|
||||
|
||||
/** Ceiling on remembered subscription-id purposes. See [subPurpose]. */
|
||||
private const val MAX_TRACKED_SUBS = 4_000
|
||||
|
||||
/** Recent-event-id window. See [recentEventIds]. */
|
||||
private const val MAX_TRACKED_EVENTS = 50_000
|
||||
}
|
||||
}
|
||||
|
||||
+12
-2
@@ -38,7 +38,9 @@ import java.util.concurrent.atomic.AtomicLong
|
||||
* hands off the accumulated value atomically, whereas remove+sum on a
|
||||
* LongAdder can strand a racing increment on an orphaned cell. Entries stay
|
||||
* in the map after a drain — the key space is small and fixed (dims x areas),
|
||||
* so this costs a few hundred boxed zeros at most.
|
||||
* so this costs a few hundred boxed zeros at most. The churn counters roughly
|
||||
* tripled it, but every one of them is still compile-time bounded: no counter
|
||||
* is keyed on a relay url, a host, or any other runtime string.
|
||||
*
|
||||
* Counters added from inside a pre-flush hook (the CPU sampler, the segment
|
||||
* integrators closing an open segment) never re-arm the debounce: they are
|
||||
@@ -74,8 +76,16 @@ class ResourceUsageAccountant(
|
||||
amount: Long,
|
||||
) {
|
||||
if (amount <= 0) return
|
||||
live.computeIfAbsent(key) { AtomicLong() }.addAndGet(amount)
|
||||
// Plain get first: computeIfAbsent locks the bin head when the key is present
|
||||
// but not the head node, and the churn counters roughly tripled the key count
|
||||
// (so collisions) on a path that runs per relay frame.
|
||||
(live[key] ?: live.computeIfAbsent(key) { AtomicLong() }).addAndGet(amount)
|
||||
if (inHookRun.get() == true) return
|
||||
// Plain read before the CAS: in steady state a flush is always already armed,
|
||||
// and the churn counters made this 5-8 calls per relay frame — every one of
|
||||
// which would otherwise be a read-modify-write on the same shared cache line
|
||||
// from every relay's socket thread, only to fail.
|
||||
if (flushScheduled.get()) return
|
||||
if (flushScheduled.compareAndSet(false, true)) {
|
||||
scope.launch {
|
||||
delay(flushDebounceMs)
|
||||
|
||||
+53
-1
@@ -75,16 +75,49 @@ class ResourceUsageReportAssembler {
|
||||
|
||||
sb.append("\nTechnical details (per epoch-day):\n")
|
||||
sb.append("```\n")
|
||||
days.toSortedMap().forEach { (day, counters) ->
|
||||
val sorted = days.toSortedMap()
|
||||
val included = newestDaysWithin(sorted, MAX_DUMP_CHARS)
|
||||
sorted.forEach { (day, counters) ->
|
||||
if (day !in included) return@forEach
|
||||
sb.append("day $day (today=$today)\n")
|
||||
counters.toSortedMap().forEach { (key, value) ->
|
||||
sb.append(" $key = $value\n")
|
||||
}
|
||||
}
|
||||
val omitted = sorted.size - included.size
|
||||
if (omitted > 0) {
|
||||
sb.append("($omitted earlier day(s) omitted to keep this report sendable; ")
|
||||
sb.append("the summary tables above still cover them)\n")
|
||||
}
|
||||
sb.append("```\n")
|
||||
return sb.toString()
|
||||
}
|
||||
|
||||
/**
|
||||
* The most recent days whose dumps fit in [budget], newest first, always
|
||||
* including at least the newest even if it alone exceeds it.
|
||||
*
|
||||
* Bounded by size rather than by a day count because the per-day size is not a
|
||||
* constant: it tracks how many distinct counters the build emits, and that has
|
||||
* grown by more than an order of magnitude. A fixed day count would have to be
|
||||
* re-tuned every time a counter family is added, and would be wrong in the
|
||||
* meantime.
|
||||
*/
|
||||
private fun newestDaysWithin(
|
||||
days: Map<Long, Map<String, Long>>,
|
||||
budget: Int,
|
||||
): Set<Long> {
|
||||
val included = mutableSetOf<Long>()
|
||||
var left = budget
|
||||
for (day in days.keys.sortedDescending()) {
|
||||
val size = days.getValue(day).entries.sumOf { it.key.length + DUMP_LINE_OVERHEAD }
|
||||
if (included.isNotEmpty() && size > left) break
|
||||
included.add(day)
|
||||
left -= size
|
||||
}
|
||||
return included
|
||||
}
|
||||
|
||||
private fun summaryTable(s: UsageSummary): String =
|
||||
buildString {
|
||||
append("| Metric | Value |\n")
|
||||
@@ -132,6 +165,25 @@ class ResourceUsageReportAssembler {
|
||||
/** Markdown table header/body separator row. */
|
||||
private const val TABLE_SEPARATOR = "| --- | --- |\n"
|
||||
|
||||
/**
|
||||
* Character budget for the raw per-day dump.
|
||||
*
|
||||
* This report exists to be sent to the developers as a NIP-17 DM, and relays
|
||||
* commonly cap events between 64 and 256 KB — so an unbounded dump does not
|
||||
* merely inconvenience, it makes the report unsendable by exactly the users
|
||||
* whose ledgers are most worth seeing. It also travels through a ~1 MB Binder
|
||||
* transaction when shared.
|
||||
*
|
||||
* The ledger keeps 30 days and a busy day now emits ~1,200 counters, which is
|
||||
* ~52 KB of dump per day — so "every retained day" would be ~1.5 MB. This
|
||||
* keeps the newest days and says how many it dropped; the summary tables
|
||||
* above are unaffected and still cover the whole window.
|
||||
*/
|
||||
private const val MAX_DUMP_CHARS = 64 * 1024
|
||||
|
||||
/** ` ` + ` = ` + the value, per dumped line. */
|
||||
private const val DUMP_LINE_OVERHEAD = 24
|
||||
|
||||
fun formatBytes(bytes: Long): String =
|
||||
when {
|
||||
bytes >= 1024L * 1024L * 1024L -> String.format(Locale.US, "%.2f GB", bytes / (1024.0 * 1024.0 * 1024.0))
|
||||
|
||||
+20
-2
@@ -34,13 +34,31 @@ import java.io.File
|
||||
* Jackson + Mutex + write-to-tmp-then-rename + version envelope.
|
||||
*
|
||||
* Day keys are UTC epoch-days (stringified for JSON). Buckets older than
|
||||
* [keepDays] are pruned on every merge, so the file stays small (a few KB).
|
||||
* [keepDays] are pruned on every merge, so the file stays small (a few KB, on a
|
||||
* key space the churn counters tripled but left compile-time bounded). The whole
|
||||
* file is re-serialized on every flush (debounced to ~30s while traffic flows),
|
||||
* so that size is paid on each write, not just at rest.
|
||||
* Also carries the high-consumption alert state (last prompt time, opt-out)
|
||||
* so the whole feature has exactly one file.
|
||||
*/
|
||||
class ResourceUsageStore(
|
||||
private val storageFile: File,
|
||||
private val keepDays: Long = 30,
|
||||
/**
|
||||
* Retention, in days.
|
||||
*
|
||||
* Seven because that is the widest window anything reads: the summary tables and
|
||||
* the trend chart both span `today - 6 .. today`, the alert evaluator looks at
|
||||
* two days, and the report's raw dump is byte-bounded well below a week. At 30 —
|
||||
* the previous value — twenty-three days were rewritten on every flush and read
|
||||
* by nothing.
|
||||
*
|
||||
* That is not free: [persist] rewrites the whole file on every merge, and the
|
||||
* relay-churn counters took a day's bucket from ~57 keys to ~241. Retention is
|
||||
* therefore a write-amplification setting as much as a history setting — cutting
|
||||
* it to a week is what keeps those counters at ~18% over the previous file size
|
||||
* rather than ~5x.
|
||||
*/
|
||||
private val keepDays: Long = 7,
|
||||
) {
|
||||
data class UsageFile(
|
||||
val version: Int = 1,
|
||||
|
||||
+629
-16
@@ -20,6 +20,18 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.service.resourceusage
|
||||
|
||||
import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.SubPurpose
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix.AUTH_REQUIRED
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix.BLOCKED
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix.ERROR
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix.INVALID
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix.RATE_LIMITED
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix.RESTRICTED
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix.UNSUPPORTED
|
||||
import com.vitorpamplona.quartz.nip66RelayMonitor.reachability.RelayObserver
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* Counter-key grammar for the resource-usage ledger. Keys are flat strings so
|
||||
* the on-disk store is schema-free — adding a counter never needs a migration.
|
||||
@@ -29,8 +41,28 @@ package com.vitorpamplona.amethyst.service.resourceusage
|
||||
* - visibility: `fg` (an activity is started) vs `bg`
|
||||
* - direction: `rx` (downloaded) vs `tx` (uploaded)
|
||||
*
|
||||
* Counters are sizes, durations, and counts only — never URLs, relay names, or
|
||||
* content. See plans/2026-07-12-resource-usage-ledger.md.
|
||||
* Counters are sizes, durations, and counts only — never content, and never a
|
||||
* full URL, and never a relay name.
|
||||
* See plans/2026-07-12-resource-usage-ledger.md.
|
||||
*
|
||||
* ## Reserved segments — read before adding a counter
|
||||
*
|
||||
* [sumMatching] matches by dot-segment *membership*, not by prefix, and every
|
||||
* headline figure in [UsageSummary] is built from it. A new key that happens to
|
||||
* contain one of these segments silently joins that sum:
|
||||
*
|
||||
* rx tx msg connms connects connfails reqs bursts activems
|
||||
* worker runs + every value in [HTTP_ROLES]
|
||||
*
|
||||
* Concretely: a key named `relay.rx.event.mobile.bg` would be counted by
|
||||
* `traffic(MOBILE, BG)` *in addition to* `relay.msg.mobile.bg.rx`, doubling the
|
||||
* reported data usage and halving the effective threshold of the
|
||||
* background-mobile-data alert. That is why the relay verb split below uses
|
||||
* `up`/`down` rather than `tx`/`rx`.
|
||||
*
|
||||
* `ResourceUsageLedgerTest.newKeysDoNotDisturbSummary` is the regression guard:
|
||||
* it asserts [UsageSummary.from] is value-identical with and without every key
|
||||
* this object can produce.
|
||||
*/
|
||||
object UsageKeys {
|
||||
const val MOBILE = "mobile"
|
||||
@@ -54,6 +86,42 @@ object UsageKeys {
|
||||
|
||||
val HTTP_ROLES = listOf(ROLE_IMAGE, ROLE_VIDEO, ROLE_UPLOADS, ROLE_MONEY, ROLE_NIP05, ROLE_PREVIEW, ROLE_PUSH, ROLE_OTHER)
|
||||
|
||||
/**
|
||||
* `mobile.bg` — the network x visibility pair every counter is split by.
|
||||
*
|
||||
* Table-backed rather than interpolated: this is evaluated on every relay frame
|
||||
* and every HTTP response, and there are only four possible answers. Declared
|
||||
* first because the key tables below are built from it at class-init.
|
||||
*/
|
||||
fun dim(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = DIMS[dimIndex(mobile, foreground)]
|
||||
|
||||
private val DIMS = arrayOf("$WIFI.$BG", "$WIFI.$FG", "$MOBILE.$BG", "$MOBILE.$FG")
|
||||
|
||||
private fun dimIndex(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): Int = (if (mobile) 2 else 0) or (if (foreground) 1 else 0)
|
||||
|
||||
/**
|
||||
* The four `<prefix>.<dim>[.<suffix>]` keys, indexed by [dimIndex].
|
||||
*
|
||||
* Used for every `relay.*` key, because all of them are built from a relay
|
||||
* callback — per frame for [relayMsg]/[relayVerb], per dial/connect/disconnect
|
||||
* for the rest. The `net.*` builders below still interpolate: their key space is
|
||||
* role x dim x metric and they are called once per HTTP response, so the table
|
||||
* would be larger and buy less. If you add a `relay.*` counter, table it.
|
||||
*/
|
||||
private fun dimKeys(
|
||||
prefix: String,
|
||||
suffix: String? = null,
|
||||
): Array<String> {
|
||||
val tail = if (suffix == null) "" else ".$suffix"
|
||||
return Array(DIMS.size) { "$prefix.${DIMS[it]}$tail" }
|
||||
}
|
||||
|
||||
/** `net.image.mobile.bg.rx` — HTTP bytes for a subsystem. */
|
||||
fun net(
|
||||
role: String,
|
||||
@@ -87,25 +155,556 @@ object UsageKeys {
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
received: Boolean,
|
||||
): String = "relay.msg.${dim(mobile, foreground)}.${if (received) RX else TX}"
|
||||
): String = (if (received) RELAY_MSG_RX else RELAY_MSG_TX)[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_MSG_RX = dimKeys("relay.msg", RX)
|
||||
private val RELAY_MSG_TX = dimKeys("relay.msg", TX)
|
||||
|
||||
/** `relay.connms.mobile.bg` — Σ(open relay connections × elapsed ms). */
|
||||
fun relayConnMs(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = "relay.connms.${dim(mobile, foreground)}"
|
||||
): String = RELAY_CONNMS[dimIndex(mobile, foreground)]
|
||||
|
||||
/** `relay.connects.mobile.bg` — completed relay (re)connections: each one paid a TCP+TLS handshake. */
|
||||
private val RELAY_CONNMS = dimKeys("relay.connms")
|
||||
|
||||
/**
|
||||
* `relay.connects.mobile.bg` — completed relay (re)connections: each one paid a
|
||||
* TCP+TLS handshake.
|
||||
*
|
||||
* Also the [relayLife] histogram's denominator — one session begins per
|
||||
* `onConnected` — which is what makes the histogram's deficit measurable rather
|
||||
* than assumed:
|
||||
*
|
||||
* connects − Σ life buckets − orphan = still open at report time + lost to process death
|
||||
*
|
||||
* Without that subtraction a leak, a still-open session and a session lost to a
|
||||
* background kill are indistinguishable.
|
||||
*/
|
||||
fun relayConnects(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = "relay.connects.${dim(mobile, foreground)}"
|
||||
): String = RELAY_CONNECTS[dimIndex(mobile, foreground)]
|
||||
|
||||
/** `relay.connfails.mobile.bg` — dials that failed before the websocket opened. */
|
||||
private val RELAY_CONNECTS = dimKeys("relay.connects")
|
||||
|
||||
/**
|
||||
* `relay.connfails.mobile.bg` — every `onCannotConnect`.
|
||||
*
|
||||
* NOT a failed-dial count, despite the name. `BasicRelayClient.onFailure`
|
||||
* raises `onCannotConnect` with no `isReady` test, so a connection that lived
|
||||
* for ten minutes and then dropped increments both this and [relayConnects]
|
||||
* from a single dial. Use [relayDials] for the actual number of dials.
|
||||
*/
|
||||
fun relayConnectFails(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = "relay.connfails.${dim(mobile, foreground)}"
|
||||
): String = RELAY_CONNFAILS[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_CONNFAILS = dimKeys("relay.connfails")
|
||||
|
||||
/**
|
||||
* `relay.dials.mobile.bg` — dial attempts, from `onConnecting`.
|
||||
*
|
||||
* Fires exactly once per dial, after the transport gate and the connect mutex
|
||||
* and before the socket is built, so this is the honest denominator that
|
||||
* [relayConnectFails] is not.
|
||||
*/
|
||||
fun relayDials(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_DIALS[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_DIALS = dimKeys("relay.dials")
|
||||
|
||||
/** `relay.disc.mobile.bg` — every `onDisconnected`, whatever the cause. */
|
||||
fun relayDisconnects(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_DISC[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_DISC = dimKeys("relay.disc")
|
||||
|
||||
/**
|
||||
* `relay.subs.sent.mobile.bg` — REQ commands sent.
|
||||
*
|
||||
* A count to sit beside the `relay.verb.up.req` byte total: PR #3832 raised the
|
||||
* number of live subscriptions per relay (background accounts now subscribe too)
|
||||
* and measured refusals against nos.lol's cap of 20. Divided by [relayConnects]
|
||||
* this is REQs per connection, which is what separates "we reconnect too often"
|
||||
* from "each reconnect asks for too much" — different fixes.
|
||||
*/
|
||||
fun relaySubsSent(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_SUBS_SENT[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_SUBS_SENT = dimKeys("relay.subs.sent")
|
||||
|
||||
/** `relay.subs.closed.mobile.bg` — CLOSE commands sent; sent minus closed is net subscription growth. */
|
||||
fun relaySubsClosed(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_SUBS_CLOSED[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_SUBS_CLOSED = dimKeys("relay.subs.closed")
|
||||
|
||||
/**
|
||||
* `relay.subs.replay.mobile.bg` — REQs sent within [REPLAY_WINDOW_MS] of that
|
||||
* relay's connect, i.e. the post-connect resubscribe burst.
|
||||
*
|
||||
* An estimate, not an exact split. `PoolRequests.syncState` replays every desired
|
||||
* filter from a coroutine launched at `onConnected`, but nothing on the listener
|
||||
* side marks a frame as belonging to it, so this is a time window. It is the
|
||||
* measurement behind the source report's inference that ~24 KB per connection
|
||||
* "is exactly the size of a full REQ subscription replay" — which was arithmetic
|
||||
* on a daily total, not an observation.
|
||||
*/
|
||||
fun relaySubsReplay(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_SUBS_REPLAY[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_SUBS_REPLAY = dimKeys("relay.subs.replay")
|
||||
|
||||
/** How long after a connect a REQ still counts as part of the resubscribe burst. */
|
||||
const val REPLAY_WINDOW_MS = 2_000L
|
||||
|
||||
/**
|
||||
* `relay.notice.toomanysubs` — NOTICE frames by reason.
|
||||
*
|
||||
* Exists because a refused subscription is otherwise invisible. Per PR #3832's
|
||||
* own known issue, `ERROR: too many concurrent REQs` arrives as a NOTICE, which
|
||||
* carries no subscription id and so never reaches `RelayReqRefusals.onRefused`
|
||||
* (wired to CLOSED only). The relay drops the REQ while the client still believes
|
||||
* it is live — it never EOSEs, its `since` never advances, and `syncState` then
|
||||
* re-requests its full backlog on every reconnect, forever. A non-trivial count
|
||||
* here means subscription pressure is a *cause* of the download volume rather
|
||||
* than a symptom of the reconnect count.
|
||||
*
|
||||
* The reason comes from a fixed allowlist, never from the relay's text. Relay
|
||||
* prose is server-controlled and this key is persisted for 30 days; `RelayObserver`
|
||||
* had to fix exactly this bug, where free-form CLOSED prose became its own tally
|
||||
* key and cardinality grew with the number of distinct sentences relays wrote.
|
||||
*/
|
||||
fun relayNotice(reason: String): String = RELAY_NOTICE[reason] ?: RELAY_NOTICE.getValue(NOTICE_UNCLASSIFIED)
|
||||
|
||||
const val NOTICE_TOO_MANY_SUBS = "toomanysubs"
|
||||
const val NOTICE_RATE_LIMITED = "ratelimited"
|
||||
const val NOTICE_AUTH_REQUIRED = "authrequired"
|
||||
const val NOTICE_RESTRICTED = "restricted"
|
||||
const val NOTICE_INVALID = "invalid"
|
||||
const val NOTICE_BLOCKED = "blocked"
|
||||
const val NOTICE_ERROR = "error"
|
||||
const val NOTICE_UNSUPPORTED = "unsupported"
|
||||
|
||||
/** The query itself was too expensive — too many kinds/steps/filters. Observed on nostr.land, relay.layer.systems. */
|
||||
const val NOTICE_QUERY_COST = "querycost"
|
||||
|
||||
/** The relay refuses REQs outright. Observed on sendit.nosflare.com. */
|
||||
const val NOTICE_REQ_REFUSED = "reqrefused"
|
||||
|
||||
/** Relay chatter that costs bytes but means nothing — keepalives, per-query PERF telemetry. */
|
||||
const val NOTICE_BENIGN = "benign"
|
||||
|
||||
/** Deliberately not `other`: that is an [HTTP_ROLES] value and a reserved segment. */
|
||||
const val NOTICE_UNCLASSIFIED = "unclassified"
|
||||
|
||||
val NOTICE_REASONS =
|
||||
listOf(
|
||||
NOTICE_TOO_MANY_SUBS,
|
||||
NOTICE_RATE_LIMITED,
|
||||
NOTICE_AUTH_REQUIRED,
|
||||
NOTICE_RESTRICTED,
|
||||
NOTICE_INVALID,
|
||||
NOTICE_BLOCKED,
|
||||
NOTICE_ERROR,
|
||||
NOTICE_UNSUPPORTED,
|
||||
NOTICE_QUERY_COST,
|
||||
NOTICE_REQ_REFUSED,
|
||||
NOTICE_BENIGN,
|
||||
NOTICE_UNCLASSIFIED,
|
||||
)
|
||||
|
||||
private val RELAY_NOTICE = NOTICE_REASONS.associateWith { "relay.notice.$it" }
|
||||
|
||||
/**
|
||||
* Classifies a NOTICE into one of [NOTICE_REASONS].
|
||||
*
|
||||
* Matches on content markers rather than the NIP-01 machine-readable prefix
|
||||
* alone, because the case this exists for does not have a useful one: strfry
|
||||
* sends `ERROR: too many concurrent REQs`, whose prefix is just `error`.
|
||||
* [RelayObserver.prefixOf] is consulted for the standard prefixes it does
|
||||
* handle correctly.
|
||||
*/
|
||||
fun noticeReason(message: String): String {
|
||||
val text = message.lowercase()
|
||||
// Several relays prefix a NOTICE with the subscription id it concerns
|
||||
// ("Kgo0HH: closed: too many steps"), which makes the *subscription id* the
|
||||
// machine-readable prefix and hides the real one. Try the remainder too.
|
||||
val prefix = RelayObserver.prefixOf(message)
|
||||
val inner = RelayObserver.prefixOf(message.substringAfter(':', ""))
|
||||
val prefixes = setOf(prefix, inner)
|
||||
return when {
|
||||
"too many" in text && ("req" in text || "subscription" in text || "concurrent" in text) -> NOTICE_TOO_MANY_SUBS
|
||||
// A cost refusal is still a refusal: the REQ is dropped, so it never
|
||||
// EOSEs and its `since` never advances.
|
||||
"too many" in text || "too costly" in text || "too expensive" in text -> NOTICE_QUERY_COST
|
||||
"does not accept" in text || "denied" in text || "not accepting" in text -> NOTICE_REQ_REFUSED
|
||||
"keepalive" in text || "perf:" in text -> NOTICE_BENIGN
|
||||
// Wire codes come from the enum rather than being hand-written a third
|
||||
// time (after the enum itself and the NOTICE_* key segments).
|
||||
RATE_LIMITED.code in prefixes || ("rate" in text && "limit" in text) -> NOTICE_RATE_LIMITED
|
||||
AUTH_REQUIRED.code in prefixes || ("auth" in text && "required" in text) -> NOTICE_AUTH_REQUIRED
|
||||
RESTRICTED.code in prefixes -> NOTICE_RESTRICTED
|
||||
INVALID.code in prefixes -> NOTICE_INVALID
|
||||
BLOCKED.code in prefixes -> NOTICE_BLOCKED
|
||||
UNSUPPORTED.code in prefixes -> NOTICE_UNSUPPORTED
|
||||
ERROR.code in prefixes -> NOTICE_ERROR
|
||||
else -> NOTICE_UNCLASSIFIED
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The bar that decides reconnect behaviour, and so also what counts as a short
|
||||
* session: below it a disconnect keeps the growing backoff,
|
||||
* at or above it the backoff resets to 1s. (`NostrClient.KEEP_ALIVE_INTERVAL_MS`
|
||||
* is the same 60s, but it is private, so this reads the one that is public.)
|
||||
*
|
||||
* Derived rather than copied: the plan retunes `STABLE_CONNECTION_IN_SECS` once
|
||||
* the histogram is read, and a hand-written 60_000 here would silently stop
|
||||
* meaning "session that kept the backoff growing" at that point.
|
||||
* `UsageKeyHelpersTest.lifeBucketsAreHalfOpen` asserts the resulting bucket
|
||||
* labels, so a retune surfaces as a test failure rather than as a histogram that
|
||||
* quietly answers the wrong question.
|
||||
*/
|
||||
const val SHORT_SESSION_MS = BasicRelayClient.STABLE_CONNECTION_IN_SECS * 1_000L
|
||||
|
||||
/**
|
||||
* A half-open millisecond histogram: ascending upper [bounds], the derived
|
||||
* bucket labels, and the `<prefix>.<bucket>.<dim>` key table they index.
|
||||
*
|
||||
* Shared by all three histograms below (`relay.life`, `relay.hs`, `relay.gap`),
|
||||
* which differ only in their bounds and in whether the label reads in seconds or
|
||||
* milliseconds. Deriving the names from the bounds is what keeps a bound change
|
||||
* from leaving a label lying about it.
|
||||
*/
|
||||
private class MsHistogram(
|
||||
prefix: String,
|
||||
private val bounds: LongArray,
|
||||
label: (Long) -> String,
|
||||
) {
|
||||
init {
|
||||
// [indexOf] linear-scans for the first bound greater than the elapsed
|
||||
// time, so the bounds must ascend. One of relay.life's is derived from
|
||||
// BasicRelayClient.STABLE_CONNECTION_IN_SECS, and raising that to five
|
||||
// minutes — exactly the retune commit 2 of the churn plan contemplates —
|
||||
// would push it past the two bounds after it. The buckets between would
|
||||
// become unreachable and the labels would start lying. Fail at class-init
|
||||
// with the reason rather than as a puzzling boundary-test failure.
|
||||
require(bounds.asList() == bounds.sorted()) {
|
||||
"$prefix bounds must ascend, got ${bounds.toList()}. " +
|
||||
"SHORT_SESSION_MS is ${SHORT_SESSION_MS}ms — reorder the bounds to match."
|
||||
}
|
||||
}
|
||||
|
||||
val names = Array(bounds.size + 1) { i -> if (i < bounds.size) "lt${label(bounds[i])}" else "gte${label(bounds.last())}" }
|
||||
|
||||
private val keys = Array(names.size) { dimKeys("$prefix.${names[it]}") }
|
||||
|
||||
fun indexOf(ms: Long): Int {
|
||||
for (i in bounds.indices) {
|
||||
if (ms < bounds[i]) return i
|
||||
}
|
||||
return bounds.size
|
||||
}
|
||||
|
||||
fun nameOf(ms: Long): String = names[indexOf(ms)]
|
||||
|
||||
fun key(
|
||||
ms: Long,
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = keys[indexOf(ms)][dimIndex(mobile, foreground)]
|
||||
}
|
||||
|
||||
/**
|
||||
* Bounds for the [relayLife] histogram deliberately straddle [SHORT_SESSION_MS]:
|
||||
* a mean cannot tell a tight cluster sitting on that bar from a bimodal mix;
|
||||
* this can.
|
||||
*/
|
||||
private val LIFE = MsHistogram("relay.life", longArrayOf(5_000, 30_000, SHORT_SESSION_MS, 120_000, 300_000)) { "${it / 1000}s" }
|
||||
|
||||
fun lifeBucket(elapsedMs: Long): String = LIFE.nameOf(elapsedMs)
|
||||
|
||||
/**
|
||||
* `relay.life.lt60s.mobile.bg` — connections that closed after living this long.
|
||||
* [relayConnects] is the denominator; see its doc for the deficit equation.
|
||||
*/
|
||||
fun relayLife(
|
||||
elapsedMs: Long,
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = LIFE.key(elapsedMs, mobile, foreground)
|
||||
|
||||
/**
|
||||
* `relay.life.overwrite` — a connect arrived for a relay that already had an
|
||||
* unconsumed start stamp.
|
||||
*
|
||||
* Expected during a pool teardown: `NostrClient` runs `disconnect()` then
|
||||
* `connect()` synchronously, so a stale failure callback for the old socket
|
||||
* can land after the new socket is already open. The new stamp overwrites the
|
||||
* old, and the stale disconnect then consumes the new one — booking a
|
||||
* near-zero lifetime for a session that never ended. Bias runs toward `lt5s`,
|
||||
* so read this before reading the histogram's short buckets.
|
||||
*/
|
||||
const val RELAY_LIFE_OVERWRITE = "relay.life.overwrite"
|
||||
|
||||
/** `relay.life.orphan` — a disconnect with no matching start stamp. */
|
||||
const val RELAY_LIFE_ORPHAN = "relay.life.orphan"
|
||||
|
||||
/**
|
||||
* `relay.verb.up.req.mobile.bg` / `relay.verb.down.eose.mobile.bg` — the
|
||||
* [relayMsg] bytes, split by wire verb. Parameterised on direction for the same
|
||||
* reason [relayMsg] is: one memo strategy, not two.
|
||||
*
|
||||
* `verb` is the command's own `label()` (`REQ`, `EVENT`, ...) lowercased, so a
|
||||
* new `Command`/`Message` subtype maps itself and nothing can land in a
|
||||
* catch-all bucket unnoticed. Deliberately `up`/`down` rather than `tx`/`rx` —
|
||||
* see the reserved-segment note above.
|
||||
*
|
||||
* Keys are memoized because this is on the per-frame path: the verb x dim space
|
||||
* is a handful of entries, so steady state is a map lookup and an array index
|
||||
* with no string building at all.
|
||||
*/
|
||||
fun relayVerb(
|
||||
verb: String,
|
||||
received: Boolean,
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String =
|
||||
(if (received) VERB_DOWN_KEYS else VERB_UP_KEYS)
|
||||
.getOrPut(verb) { dimKeys("relay.verb.${if (received) DOWN else UP}.${verb.lowercase()}") }[dimIndex(mobile, foreground)]
|
||||
|
||||
private const val UP = "up"
|
||||
private const val DOWN = "down"
|
||||
|
||||
private val VERB_UP_KEYS = ConcurrentHashMap<String, Array<String>>()
|
||||
private val VERB_DOWN_KEYS = ConcurrentHashMap<String, Array<String>>()
|
||||
|
||||
/**
|
||||
* `relay.purpose.home_feed.sent` / `.bytes` — REQ frames and REQ bytes by the
|
||||
* [SubPurpose] that asked for them.
|
||||
*
|
||||
* The counter that turns "REQ traffic is 64 % of upload" into an actionable
|
||||
* name. Purpose travels on the filter itself (PR #3832's `ExplainedFilter`,
|
||||
* which survives the `copy(since = …)` assemblers do after every EOSE), so this
|
||||
* is a read, not a new registry.
|
||||
*
|
||||
* Cardinality is the enum, so it is bounded and stable. [PURPOSE_UNEXPLAINED] is
|
||||
* its own bucket rather than folded into the enum's OTHER: a filter carrying no
|
||||
* purpose at all means an assembler #3832 did not reach, which is a different
|
||||
* fact from one that declared itself uncategorised — and if that bucket is large,
|
||||
* the attribution below cannot be trusted.
|
||||
*
|
||||
* Memoized for the same reason [relayVerb] is, and more urgently: [relayPurposeDown]
|
||||
* and [relayPurposeDownCount] both run on every inbound frame that names a
|
||||
* subscription, so interpolating would build two strings per frame on the hottest
|
||||
* path in the app. The purpose space is the enum plus the three fallbacks below.
|
||||
*/
|
||||
fun relayPurposeSent(purpose: String): String = purposeKeys(purpose)[P_SENT]
|
||||
|
||||
fun relayPurposeBytes(purpose: String): String = purposeKeys(purpose)[P_BYTES]
|
||||
|
||||
/**
|
||||
* `relay.purpose.moderation.down` — bytes received on subscriptions opened for
|
||||
* that purpose, resolved through the subscription id the frame carries.
|
||||
*
|
||||
* The upload counters answer "who is asking"; this answers "who is being
|
||||
* answered", which is the larger number: inbound EVENT payload is ~74 % of relay
|
||||
* traffic against ~26 % outbound. Without it, a fix to the REQ churn can only be
|
||||
* credited with the upload it removes, when the interesting question is how much
|
||||
* of the download it was causing — every re-subscription can make the relay
|
||||
* re-send everything that matches.
|
||||
*/
|
||||
fun relayPurposeDown(purpose: String): String = purposeKeys(purpose)[P_DOWN]
|
||||
|
||||
/**
|
||||
* `relay.purpose.home_feed.downn` — inbound frames, alongside the bytes.
|
||||
*
|
||||
* Bytes alone cannot separate "many small events delivered repeatedly" from "few
|
||||
* large ones", and those want opposite fixes. With a count, `down / downn` is the
|
||||
* average frame size per purpose, and the total frame count set against
|
||||
* `crypto.verify.count` — which the cache pays once per event it accepts — bounds
|
||||
* how much of the download is the same events arriving from different relays
|
||||
* under the outbox fan-out.
|
||||
*/
|
||||
fun relayPurposeDownCount(purpose: String): String = purposeKeys(purpose)[P_DOWNN]
|
||||
|
||||
/**
|
||||
* `relay.purpose.user_profile.dupbytes` — of that purpose's inbound bytes, how
|
||||
* many carried an event already delivered.
|
||||
*
|
||||
* [relayEventsDupBytes] measures duplication across the whole client, which says
|
||||
* how much is wasted but not where. The seventh reading needs exactly this split:
|
||||
* `user_profile` was 57 % of download at ~17 KB per event, and whether that is
|
||||
* mostly the same events arriving from many relays or mostly distinct large ones
|
||||
* points at completely different fixes — suppress redundant delivery, or stop
|
||||
* fetching the large thing per relay.
|
||||
*/
|
||||
fun relayPurposeDupBytes(purpose: String): String = purposeKeys(purpose)[P_DUPBYTES]
|
||||
|
||||
private const val P_SENT = 0
|
||||
private const val P_BYTES = 1
|
||||
private const val P_DOWN = 2
|
||||
private const val P_DOWNN = 3
|
||||
private const val P_DUPBYTES = 4
|
||||
|
||||
private val PURPOSE_SUFFIXES = arrayOf("sent", "bytes", "down", "downn", "dupbytes")
|
||||
|
||||
private val PURPOSE_KEYS = ConcurrentHashMap<String, Array<String>>()
|
||||
|
||||
/** The five `relay.purpose.<purpose>.*` keys, indexed by the `P_` constants above. */
|
||||
private fun purposeKeys(purpose: String): Array<String> = PURPOSE_KEYS.getOrPut(purpose) { Array(PURPOSE_SUFFIXES.size) { "relay.purpose.$purpose.${PURPOSE_SUFFIXES[it]}" } }
|
||||
|
||||
/** A frame whose subscription id we never saw opened — counters wiped mid-session, or a sub from before this connection. */
|
||||
const val PURPOSE_UNATTRIBUTED = "unattributed"
|
||||
|
||||
/** A REQ whose filters carry no [ExplainedFilter] purpose. */
|
||||
const val PURPOSE_UNEXPLAINED = "unexplained"
|
||||
|
||||
/** The enum's own OTHER, renamed: bare `other` is an [HTTP_ROLES] value and a reserved segment. */
|
||||
const val PURPOSE_OTHER = "otherpurpose"
|
||||
|
||||
/**
|
||||
* The key segment for a [SubPurpose]. The one place the enum is turned into a
|
||||
* key, so the reserved-segment rename cannot drift between the producer and the
|
||||
* test that guards it — which is exactly how it drifted the first time.
|
||||
*
|
||||
* Tabled by ordinal: this runs per REQ, and `name.lowercase()` would allocate a
|
||||
* fresh String each time for one of a dozen fixed answers.
|
||||
*/
|
||||
fun purposeKeyPart(purpose: SubPurpose): String = PURPOSE_PARTS[purpose.ordinal]
|
||||
|
||||
private val PURPOSE_PARTS =
|
||||
Array(SubPurpose.entries.size) {
|
||||
val purpose = SubPurpose.entries[it]
|
||||
if (purpose == SubPurpose.OTHER) PURPOSE_OTHER else purpose.name.lowercase()
|
||||
}
|
||||
|
||||
/**
|
||||
* `relay.subs.resent.mobile.bg` — a REQ for a subscription id this relay already
|
||||
* has open on the current connection.
|
||||
*
|
||||
* The distinction the churn question turns on. A REQ that opens a new
|
||||
* subscription is work; a REQ that replaces one already in flight is the client
|
||||
* changing its mind, and at ~1 KB each that is pure cost. Measured against
|
||||
* [relaySubsSent] it says what fraction of the upload is re-subscription rather
|
||||
* than subscription.
|
||||
*/
|
||||
fun relaySubsResent(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_SUBS_RESENT[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_SUBS_RESENT = dimKeys("relay.subs.resent")
|
||||
|
||||
/** Half-open bounds, in ms, shared by the two connect-timing histograms below. */
|
||||
private val CONNECT_BOUNDS_MS = longArrayOf(100, 500, 2_000, 10_000, 30_000)
|
||||
|
||||
/**
|
||||
* `relay.hs.lt500ms.wifi.fg` — the websocket upgrade round-trip, as the
|
||||
* transport measured it.
|
||||
*
|
||||
* This is `onConnected`'s `pingMillis`, which `BasicOkHttpWebSocket` computes as
|
||||
* `receivedResponseAtMillis - sentRequestAtMillis` and which this listener
|
||||
* previously discarded. PR #3843 is the cautionary tale: `RelayObserver` derived
|
||||
* the same quantity from `onConnecting -> onConnected` instead, and on a large
|
||||
* fan-out published a 33.5 s median that was the client's own backlog rather than
|
||||
* relay latency. Those timestamps bracket the request itself, so everything
|
||||
* before it — queueing, DNS, TCP, TLS — is excluded. Zero or negative means the
|
||||
* transport could not time it, and nothing is recorded rather than a fabricated 0.
|
||||
*/
|
||||
fun relayHandshake(
|
||||
ms: Long,
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = HANDSHAKE.key(ms, mobile, foreground)
|
||||
|
||||
private val HANDSHAKE = MsHistogram("relay.hs", CONNECT_BOUNDS_MS) { "${it}ms" }
|
||||
|
||||
/**
|
||||
* `relay.gap.lt2000ms.wifi.fg` — everything between deciding to dial and the
|
||||
* upgrade request going out: dispatcher queueing, DNS, TCP, TLS.
|
||||
*
|
||||
* `(onConnected wall clock - onConnecting wall clock) - handshake`. This is the
|
||||
* share of connect latency the app is responsible for rather than the relay, and
|
||||
* it is what decides whether a high never-became-ready rate is relays being
|
||||
* unreachable or ~500 simultaneous dials saturating name resolution and sockets.
|
||||
* The dispatcher's own cap is not the constraint on a phone
|
||||
* (`maxRequests = 1024`, `maxRequestsPerHost = 10`), so a large value here points
|
||||
* at resolution and socket setup, not at a queue.
|
||||
*/
|
||||
fun relayDialGap(
|
||||
ms: Long,
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = DIAL_GAP.key(ms, mobile, foreground)
|
||||
|
||||
private val DIAL_GAP = MsHistogram("relay.gap", CONNECT_BOUNDS_MS) { "${it}ms" }
|
||||
|
||||
/**
|
||||
* `relay.events.seen.wifi.fg` — inbound EVENT frames, and of those, how many
|
||||
* carried an event id already delivered recently.
|
||||
*
|
||||
* The outbox model asks many relays for the same authors, so one event is
|
||||
* delivered once per relay that carries it. Relay download is ~65 % of all data
|
||||
* on the release build, so the duplication factor decides whether the largest
|
||||
* number in the ledger is content or repetition — a question no other counter
|
||||
* here can answer, and one [VERIFY_COUNT] only proxies (it counts what the cache
|
||||
* accepted, not what arrived, and only while dedup-before-verify holds).
|
||||
*
|
||||
* [relayEventsDupBytes] is the number that matters: bytes that arrived and were
|
||||
* already held.
|
||||
*/
|
||||
fun relayEventsSeen(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_EV_SEEN[dimIndex(mobile, foreground)]
|
||||
|
||||
fun relayEventsDup(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_EV_DUP[dimIndex(mobile, foreground)]
|
||||
|
||||
fun relayEventsDupBytes(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = RELAY_EV_DUPB[dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_EV_SEEN = dimKeys("relay.events.seen")
|
||||
private val RELAY_EV_DUP = dimKeys("relay.events.dup")
|
||||
private val RELAY_EV_DUPB = dimKeys("relay.events.dupbytes")
|
||||
|
||||
/**
|
||||
* `relay.trigger.netid` — reconnect *decisions*, by cause, not reconnects
|
||||
* performed.
|
||||
*
|
||||
* Two reasons this is an upper bound, both of which matter when reading it:
|
||||
* `NostrClient.reconnect` emits into a debounced flow that `subscribe` /
|
||||
* `count` / `publish` / `onDisconnected` also feed very frequently, so a
|
||||
* teardown can be coalesced away before it runs; and the flow's initial value
|
||||
* fires one teardown per client construction with no trigger attributed.
|
||||
*/
|
||||
fun relayTrigger(cause: String): String = "relay.trigger.$cause"
|
||||
|
||||
const val TRIGGER_NETID = "netid"
|
||||
const val TRIGGER_TRANSPORT = "transport"
|
||||
const val TRIGGER_TOR_POLICY = "torpolicy"
|
||||
const val TRIGGER_CLASSIFICATION = "class"
|
||||
const val TRIGGER_COLD_START = "coldstart"
|
||||
const val TRIGGER_OFF = "off"
|
||||
const val TRIGGER_BUZZ = "buzz"
|
||||
|
||||
/** `worker.scheduledPost.runs` */
|
||||
fun workerRuns(worker: String): String = "worker.$worker.runs"
|
||||
@@ -187,18 +786,32 @@ object UsageKeys {
|
||||
const val BATTERY_DRAIN_FG = "battery.drain.fg"
|
||||
const val BATTERY_DRAIN_BG = "battery.drain.bg"
|
||||
|
||||
fun dim(
|
||||
mobile: Boolean,
|
||||
foreground: Boolean,
|
||||
): String = "${if (mobile) MOBILE else WIFI}.${if (foreground) FG else BG}"
|
||||
|
||||
/** Sums every counter whose key matches all the given dot-delimited parts. */
|
||||
/**
|
||||
* Sums every counter whose key matches all the given dot-delimited parts.
|
||||
*
|
||||
* Scans segments in place rather than `key.split('.')`: [UsageSummary.from] makes
|
||||
* ~54 of these passes over the whole day bucket, the usage screen builds 16
|
||||
* summaries per entry on the main thread, and the churn counters roughly tripled
|
||||
* the key count — none of which can ever match (that is what
|
||||
* `noNewKeyContainsAReservedSegment` guarantees), so every split was pure waste.
|
||||
*/
|
||||
fun Map<String, Long>.sumMatching(vararg parts: String): Long {
|
||||
var total = 0L
|
||||
for ((key, value) in this) {
|
||||
val segments = key.split('.')
|
||||
if (parts.all { it in segments }) total += value
|
||||
if (parts.all { key.hasSegment(it) }) total += value
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
/** True when `segment` is one of this key's whole dot-delimited segments. */
|
||||
private fun String.hasSegment(segment: String): Boolean {
|
||||
var from = 0
|
||||
while (from <= length) {
|
||||
var end = indexOf('.', from)
|
||||
if (end < 0) end = length
|
||||
if (end - from == segment.length && regionMatches(from, segment, 0, segment.length)) return true
|
||||
from = end + 1
|
||||
}
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
+1
-3
@@ -109,9 +109,7 @@ class AgentConsoleViewModel : ViewModel() {
|
||||
relay?.let {
|
||||
val newlyJoined = BuzzWorkspaces.join(it)
|
||||
viewModelScope.launch { account.relayAuthLedger.setDecision(it.url, RelayAuthDecision.ALLOW) }
|
||||
// A join makes the relay first-party; if the socket was already open its one-shot AUTH
|
||||
// challenge was spent unauthenticated, so reconnect to re-challenge and authenticate.
|
||||
if (newlyJoined) account.client.reconnect(onlyIfChanged = false, ignoreRetryDelays = true)
|
||||
if (newlyJoined) reconnectPoolAfterJoin(account.client)
|
||||
}
|
||||
refresh()
|
||||
}
|
||||
|
||||
+1
-3
@@ -130,9 +130,7 @@ class BuzzDmListViewModel : ViewModel() {
|
||||
|
||||
val newlyJoined = BuzzWorkspaces.join(relay)
|
||||
viewModelScope.launch { account.relayAuthLedger.setDecision(relay.url, RelayAuthDecision.ALLOW) }
|
||||
// A join makes the relay first-party; if the socket was already open its one-shot AUTH
|
||||
// challenge was spent unauthenticated, so reconnect to re-challenge and authenticate.
|
||||
if (newlyJoined) account.client.reconnect(onlyIfChanged = false, ignoreRetryDelays = true)
|
||||
if (newlyJoined) reconnectPoolAfterJoin(account.client)
|
||||
|
||||
// Paint from cache BEFORE any network work. [discoverMemberChannels] learns the channel ids
|
||||
// from a relay round-trip, so waiting on it left the Direct Messages section visibly empty
|
||||
|
||||
+53
@@ -0,0 +1,53 @@
|
||||
/*
|
||||
* 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.ui.screen.loggedIn.buzz
|
||||
|
||||
import com.vitorpamplona.amethyst.Amethyst
|
||||
import com.vitorpamplona.amethyst.service.resourceusage.UsageKeys
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
|
||||
|
||||
/**
|
||||
* Re-dials the whole relay pool after a Buzz workspace join.
|
||||
*
|
||||
* NIP-42 sends its AUTH challenge once, on connect. If the socket was already open
|
||||
* before the join (the common case — the relay is in the user's lists and connected
|
||||
* at startup), that challenge was spent while the relay was still NOT first-party,
|
||||
* so the connection is unauthenticated and every `#p=me`-gated read on it is
|
||||
* refused. Joining makes the relay first-party (see `AuthCoordinator.isFirstParty`);
|
||||
* this forces the relay to re-challenge so the connection authenticates.
|
||||
*
|
||||
* Shared by the three join sites that need the re-challenge, both to keep the
|
||||
* reconnect flags identical and to give the churn ledger one place to attribute from:
|
||||
* without the counter, `Σ(relay.trigger.*)` would only account for the
|
||||
* connectivity-driven teardowns `RelayProxyClientConnector` reports, and a full pool
|
||||
* teardown is the most expensive thing either can do.
|
||||
*
|
||||
* `BuzzInviteScreen` is a fourth join+pre-approve site that deliberately does NOT
|
||||
* reconnect — it hands off to the in-app browser rather than reading a `#p=me`-gated
|
||||
* subscription — so `relay.trigger.buzz` undercounts joins, not re-challenges.
|
||||
*/
|
||||
internal fun reconnectPoolAfterJoin(client: INostrClient) {
|
||||
// Guarded like every other ledger write that reaches the application singleton
|
||||
// (MediaPlayTimeTracker, the workers): a diagnostics counter must never break a
|
||||
// user-visible join, and `Amethyst.instance` is lateinit.
|
||||
runCatching { Amethyst.instance.resourceUsage.add(UsageKeys.relayTrigger(UsageKeys.TRIGGER_BUZZ), 1) }
|
||||
client.reconnect(onlyIfChanged = false, ignoreRetryDelays = true)
|
||||
}
|
||||
+2
-8
@@ -107,14 +107,8 @@ class BuzzRelayImportViewModel : ViewModel() {
|
||||
val newlyJoined = BuzzWorkspaces.join(normalized)
|
||||
viewModelScope.launch { account.relayAuthLedger.setDecision(normalized.url, RelayAuthDecision.ALLOW) }
|
||||
|
||||
// NIP-42 sends its AUTH challenge once, on connect. If the socket was already open before this
|
||||
// join (the common case — the relay is in the user's lists and connected at startup), that
|
||||
// challenge was spent while the relay was still NOT first-party, so the connection is
|
||||
// unauthenticated and the persistent group-roster (39002) subscription is refused. Joining
|
||||
// makes the relay first-party (see AuthCoordinator.isFirstParty); force a reconnect so the
|
||||
// relay re-challenges and the connection authenticates — unlocking the roster (Join gate) and
|
||||
// every other `#p=me`-gated read on the shared socket.
|
||||
if (newlyJoined) account.client.reconnect(onlyIfChanged = false, ignoreRetryDelays = true)
|
||||
// Unlocks the persistent group-roster (39002) subscription — see [reconnectPoolAfterJoin].
|
||||
if (newlyJoined) reconnectPoolAfterJoin(account.client)
|
||||
|
||||
// Track "already added" against the live kind-10009 list, scoped to this relay, so the rows
|
||||
// follow every add/remove — from here, from the channel's top bar, or from another device.
|
||||
|
||||
+66
-13
@@ -20,6 +20,7 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.ui.screen.loggedIn.settings
|
||||
|
||||
import android.content.Intent
|
||||
import androidx.annotation.StringRes
|
||||
import androidx.compose.foundation.background
|
||||
import androidx.compose.foundation.layout.Arrangement
|
||||
@@ -46,11 +47,13 @@ import androidx.compose.runtime.getValue
|
||||
import androidx.compose.runtime.mutableStateOf
|
||||
import androidx.compose.runtime.produceState
|
||||
import androidx.compose.runtime.remember
|
||||
import androidx.compose.runtime.rememberCoroutineScope
|
||||
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.graphics.Color
|
||||
import androidx.compose.ui.platform.LocalClipboard
|
||||
import androidx.compose.ui.platform.LocalContext
|
||||
import androidx.compose.ui.res.pluralStringResource
|
||||
import androidx.compose.ui.text.font.FontWeight
|
||||
@@ -67,6 +70,7 @@ import com.vitorpamplona.amethyst.service.resourceusage.ResourceUsageReportAssem
|
||||
import com.vitorpamplona.amethyst.service.resourceusage.ResourceUsageReportAssembler.Companion.formatConnHours
|
||||
import com.vitorpamplona.amethyst.service.resourceusage.ResourceUsageReportAssembler.Companion.formatDurationMs
|
||||
import com.vitorpamplona.amethyst.service.resourceusage.UsageSummary
|
||||
import com.vitorpamplona.amethyst.ui.components.util.setText
|
||||
import com.vitorpamplona.amethyst.ui.navigation.navs.INav
|
||||
import com.vitorpamplona.amethyst.ui.navigation.routes.Route
|
||||
import com.vitorpamplona.amethyst.ui.navigation.routes.routeToMessage
|
||||
@@ -77,6 +81,7 @@ import com.vitorpamplona.amethyst.ui.theme.allGoodColor
|
||||
import com.vitorpamplona.amethyst.ui.theme.warningColor
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.withContext
|
||||
import java.util.Locale
|
||||
|
||||
@@ -579,6 +584,18 @@ private fun SendReportSection(
|
||||
today: Long,
|
||||
memory: MemorySnapshot?,
|
||||
) {
|
||||
val context = LocalContext.current
|
||||
val clipboard = LocalClipboard.current
|
||||
val scope = rememberCoroutineScope()
|
||||
|
||||
// Suspending, and off Main: assembling the report walks every counter of every
|
||||
// retained day (UsageSummary makes ~54 passes per summary) over a key space the
|
||||
// churn counters grew by an order of magnitude. `scope` is Main.immediate, so a
|
||||
// bare `scope.launch { }` would still build it on the UI thread — the
|
||||
// withContext is what actually moves it. Only the handover (clipboard, chooser,
|
||||
// navigation) stays on Main.
|
||||
suspend fun buildReport(): String = withContext(Dispatchers.Default) { ResourceUsageReportAssembler().buildReport(days, today, memory) }
|
||||
|
||||
SettingsSection(R.string.resource_usage_send_section) {
|
||||
Column(
|
||||
modifier = Modifier.padding(16.dp),
|
||||
@@ -589,21 +606,57 @@ private fun SendReportSection(
|
||||
style = MaterialTheme.typography.bodyMedium,
|
||||
color = MaterialTheme.colorScheme.onSurfaceVariant,
|
||||
)
|
||||
Button(
|
||||
onClick = {
|
||||
val report = ResourceUsageReportAssembler().buildReport(days, today, memory)
|
||||
nav.nav {
|
||||
routeToMessage(
|
||||
user = LocalCache.getOrCreateUser(DEV_REPORT_PUBKEY),
|
||||
draftMessage = report,
|
||||
accountViewModel = accountViewModel,
|
||||
expiresDays = 30,
|
||||
)
|
||||
}
|
||||
},
|
||||
// The DM route needs a composer to exist before the text does, which makes
|
||||
// it a poor fit for reading the report yourself — copying out of a draft
|
||||
// message is the only way to get at it. Copy and Share hand over the same
|
||||
// string directly, for pasting into an issue or saving to a file.
|
||||
Row(
|
||||
modifier = Modifier.align(Alignment.End),
|
||||
horizontalArrangement = Arrangement.spacedBy(8.dp),
|
||||
verticalAlignment = Alignment.CenterVertically,
|
||||
) {
|
||||
Text(stringRes(R.string.resource_usage_send_button))
|
||||
TextButton(
|
||||
onClick = {
|
||||
scope.launch { clipboard.setText(buildReport()) }
|
||||
},
|
||||
) {
|
||||
Text(stringRes(R.string.resource_usage_copy_button))
|
||||
}
|
||||
TextButton(
|
||||
onClick = {
|
||||
scope.launch {
|
||||
val send =
|
||||
Intent().apply {
|
||||
action = Intent.ACTION_SEND
|
||||
type = "text/plain"
|
||||
putExtra(Intent.EXTRA_TEXT, buildReport())
|
||||
putExtra(Intent.EXTRA_TITLE, stringRes(context, R.string.resource_usage_send_section))
|
||||
}
|
||||
context.startActivity(
|
||||
Intent.createChooser(send, stringRes(context, R.string.resource_usage_share_button)),
|
||||
)
|
||||
}
|
||||
},
|
||||
) {
|
||||
Text(stringRes(R.string.resource_usage_share_button))
|
||||
}
|
||||
Button(
|
||||
onClick = {
|
||||
scope.launch {
|
||||
val report = buildReport()
|
||||
nav.nav {
|
||||
routeToMessage(
|
||||
user = LocalCache.getOrCreateUser(DEV_REPORT_PUBKEY),
|
||||
draftMessage = report,
|
||||
accountViewModel = accountViewModel,
|
||||
expiresDays = 30,
|
||||
)
|
||||
}
|
||||
}
|
||||
},
|
||||
) {
|
||||
Text(stringRes(R.string.resource_usage_send_button))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4188,8 +4188,10 @@
|
||||
<string name="resource_usage_memory_chatrooms">Chatroom lists in memory</string>
|
||||
<string name="resource_usage_memory_device_class">Device memory class</string>
|
||||
<string name="resource_usage_send_section">Share with the developers</string>
|
||||
<string name="resource_usage_send_explanation">If Amethyst seems to drain battery or data, you can send this report to the developers in an encrypted DM. It contains only the numbers on this screen and the technical counters behind them \u2014 no posts, contacts, or browsing details. Nothing is sent until you tap Send in the message screen.</string>
|
||||
<string name="resource_usage_send_explanation">If Amethyst seems to drain battery or data, you can send this report to the developers in an encrypted DM, or copy it and share it yourself. It contains only the numbers on this screen and the technical counters behind them \u2014 no posts, contacts, relay names, or browsing details. The report leaves this screen only when you send, copy, or share it.</string>
|
||||
<string name="resource_usage_send_button">Send report via DM</string>
|
||||
<string name="resource_usage_copy_button">Copy</string>
|
||||
<string name="resource_usage_share_button">Share</string>
|
||||
<string name="resource_usage_alert_title">High resource usage detected</string>
|
||||
<string name="resource_usage_alert_message">Amethyst consumed more than expected recently: %1$s. Would you like to send a usage report to the developers in an encrypted DM? You will see the full report before anything is sent.</string>
|
||||
<string name="resource_usage_reason_bg_data">%1$s of cellular data in the background in one day</string>
|
||||
|
||||
+822
@@ -20,12 +20,33 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.service.resourceusage
|
||||
|
||||
import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.ExplainedFilter
|
||||
import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.SubPurpose
|
||||
import com.vitorpamplona.amethyst.service.playback.playerPool.MediaPlayTimeTracker
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.LimitsMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NotifyMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.AuthCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CountCmd
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd
|
||||
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.signers.NostrSigner
|
||||
import com.vitorpamplona.quartz.nip57Zaps.LnZapPrivateEvent
|
||||
import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent
|
||||
import io.mockk.every
|
||||
import io.mockk.mockk
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
@@ -778,3 +799,804 @@ class ResourceUsageAlertsTest {
|
||||
assertFalse(ResourceUsageAlerts.shouldPrompt(lastAlertAtSec = 0, optOut = true, nowSec = now))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The guard that keeps the counter-key grammar honest — see the reserved-segment
|
||||
* note on [UsageKeys] for the mechanism and what it would cost to get wrong.
|
||||
*/
|
||||
class UsageKeyGrammarTest {
|
||||
/** Every diagnostic key shape, with values that would be obvious if they leaked into a sum. */
|
||||
private fun churnKeys(): Map<String, Long> {
|
||||
val out = mutableMapOf<String, Long>()
|
||||
var n = 1_000_000L
|
||||
for (mobile in listOf(true, false)) {
|
||||
for (fg in listOf(true, false)) {
|
||||
out[UsageKeys.relayDials(mobile, fg)] = n++
|
||||
out[UsageKeys.relayDisconnects(mobile, fg)] = n++
|
||||
for (ms in listOf(1L, 10_000L, 45_000L, 90_000L, 200_000L, 900_000L)) {
|
||||
out[UsageKeys.relayLife(ms, mobile, fg)] = n++
|
||||
}
|
||||
for (verb in CMD_LABELS) {
|
||||
out[UsageKeys.relayVerb(verb, received = false, mobile, fg)] = n++
|
||||
}
|
||||
for (verb in MSG_LABELS) {
|
||||
out[UsageKeys.relayVerb(verb, received = true, mobile, fg)] = n++
|
||||
}
|
||||
}
|
||||
}
|
||||
out[UsageKeys.RELAY_LIFE_OVERWRITE] = n++
|
||||
out[UsageKeys.RELAY_LIFE_ORPHAN] = n++
|
||||
for (cause in listOf(
|
||||
UsageKeys.TRIGGER_NETID,
|
||||
UsageKeys.TRIGGER_TRANSPORT,
|
||||
UsageKeys.TRIGGER_TOR_POLICY,
|
||||
UsageKeys.TRIGGER_CLASSIFICATION,
|
||||
UsageKeys.TRIGGER_COLD_START,
|
||||
UsageKeys.TRIGGER_OFF,
|
||||
UsageKeys.TRIGGER_BUZZ,
|
||||
)) {
|
||||
out[UsageKeys.relayTrigger(cause)] = n++
|
||||
}
|
||||
for (mobile in listOf(true, false)) {
|
||||
for (fg in listOf(true, false)) {
|
||||
out[UsageKeys.relaySubsSent(mobile, fg)] = n++
|
||||
out[UsageKeys.relaySubsClosed(mobile, fg)] = n++
|
||||
out[UsageKeys.relaySubsReplay(mobile, fg)] = n++
|
||||
}
|
||||
}
|
||||
UsageKeys.NOTICE_REASONS.forEach { out[UsageKeys.relayNotice(it)] = n++ }
|
||||
for (mobile in listOf(true, false)) {
|
||||
for (fg in listOf(true, false)) {
|
||||
out[UsageKeys.relaySubsResent(mobile, fg)] = n++
|
||||
out[UsageKeys.relayEventsSeen(mobile, fg)] = n++
|
||||
out[UsageKeys.relayEventsDup(mobile, fg)] = n++
|
||||
out[UsageKeys.relayEventsDupBytes(mobile, fg)] = n++
|
||||
for (ms in listOf(0L, 200L, 1_000L, 5_000L, 20_000L, 60_000L)) {
|
||||
out[UsageKeys.relayHandshake(ms, mobile, fg)] = n++
|
||||
out[UsageKeys.relayDialGap(ms, mobile, fg)] = n++
|
||||
}
|
||||
}
|
||||
}
|
||||
(SubPurpose.entries.map { UsageKeys.purposeKeyPart(it) } + UsageKeys.PURPOSE_UNEXPLAINED + UsageKeys.PURPOSE_UNATTRIBUTED).forEach {
|
||||
out[UsageKeys.relayPurposeSent(it)] = n++
|
||||
out[UsageKeys.relayPurposeBytes(it)] = n++
|
||||
out[UsageKeys.relayPurposeDown(it)] = n++
|
||||
out[UsageKeys.relayPurposeDownCount(it)] = n++
|
||||
out[UsageKeys.relayPurposeDupBytes(it)] = n++
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/** A baseline of the pre-existing counters the summary is actually built from. */
|
||||
private fun baseline(): Map<String, Long> =
|
||||
mapOf(
|
||||
UsageKeys.relayMsg(mobile = true, foreground = false, received = true) to 500L,
|
||||
UsageKeys.relayMsg(mobile = true, foreground = false, received = false) to 60L,
|
||||
UsageKeys.relayMsg(mobile = false, foreground = true, received = true) to 900L,
|
||||
UsageKeys.net(UsageKeys.ROLE_IMAGE, mobile = true, foreground = true, received = true) to 70L,
|
||||
UsageKeys.netReqs(UsageKeys.ROLE_IMAGE, mobile = true, foreground = true) to 3L,
|
||||
UsageKeys.netActiveMs(UsageKeys.ROLE_IMAGE, mobile = true, foreground = true) to 40L,
|
||||
UsageKeys.radioBursts(mobile = true, foreground = true) to 2L,
|
||||
UsageKeys.relayConnMs(mobile = true, foreground = false) to 1_234L,
|
||||
UsageKeys.relayConnects(mobile = true, foreground = false) to 11L,
|
||||
UsageKeys.relayConnectFails(mobile = true, foreground = false) to 22L,
|
||||
UsageKeys.workerRuns("calendarReminder") to 1L,
|
||||
)
|
||||
|
||||
@Test
|
||||
fun newKeysDoNotDisturbSummary() {
|
||||
val before = UsageSummary.from(baseline())
|
||||
val after = UsageSummary.from(baseline() + churnKeys())
|
||||
assertEquals(before, after)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun noNewKeyContainsAReservedSegment() {
|
||||
val reserved =
|
||||
setOf(
|
||||
UsageKeys.RX,
|
||||
UsageKeys.TX,
|
||||
"msg",
|
||||
"connms",
|
||||
"connects",
|
||||
"connfails",
|
||||
"reqs",
|
||||
"bursts",
|
||||
"activems",
|
||||
"worker",
|
||||
"runs",
|
||||
) + UsageKeys.HTTP_ROLES
|
||||
|
||||
churnKeys().keys.forEach { key ->
|
||||
val clash = key.split('.').filter { it in reserved }
|
||||
assertTrue("Key '$key' uses reserved segment(s) $clash", clash.isEmpty())
|
||||
}
|
||||
}
|
||||
|
||||
companion object {
|
||||
/**
|
||||
* The verb segments taken straight from quartz's own wire labels — the same
|
||||
* source [RelayUsageListener] reads, so a new subtype cannot be tested against
|
||||
* a stale hand-written list.
|
||||
*/
|
||||
val CMD_LABELS = listOf(ReqCmd.LABEL, EventCmd.LABEL, AuthCmd.LABEL, CloseCmd.LABEL, CountCmd.LABEL)
|
||||
val MSG_LABELS =
|
||||
listOf(
|
||||
EventMessage.LABEL,
|
||||
EoseMessage.LABEL,
|
||||
OkMessage.LABEL,
|
||||
NoticeMessage.LABEL,
|
||||
AuthMessage.LABEL,
|
||||
ClosedMessage.LABEL,
|
||||
CountMessage.LABEL,
|
||||
NotifyMessage.LABEL,
|
||||
LimitsMessage.LABEL,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
class UsageKeyHelpersTest {
|
||||
@Test
|
||||
fun lifeBucketsAreHalfOpen() {
|
||||
assertEquals("lt5s", UsageKeys.lifeBucket(0))
|
||||
assertEquals("lt5s", UsageKeys.lifeBucket(4_999))
|
||||
assertEquals("lt30s", UsageKeys.lifeBucket(5_000))
|
||||
assertEquals("lt60s", UsageKeys.lifeBucket(59_999))
|
||||
// The one that matters: exactly the stability bar is NOT "under a minute".
|
||||
assertEquals("lt120s", UsageKeys.lifeBucket(60_000))
|
||||
assertEquals("lt300s", UsageKeys.lifeBucket(299_999))
|
||||
assertEquals("gte300s", UsageKeys.lifeBucket(300_000))
|
||||
assertEquals("gte300s", UsageKeys.lifeBucket(Long.MAX_VALUE))
|
||||
}
|
||||
}
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
class RelayUsageListenerTest {
|
||||
@get:Rule val tmp = TemporaryFolder()
|
||||
|
||||
private var now = 0L
|
||||
|
||||
private fun relay(url: String): IRelayClient {
|
||||
val r = mockk<IRelayClient>(relaxed = true)
|
||||
every { r.url } returns NormalizedRelayUrl(url)
|
||||
return r
|
||||
}
|
||||
|
||||
private fun runLedger(block: suspend (ResourceUsageAccountant, RelayUsageListener) -> Unit) =
|
||||
runTest {
|
||||
val store = ResourceUsageStore(File(tmp.newFolder(), "usage.json"))
|
||||
// backgroundScope, as everywhere else in this file: the accountant's
|
||||
// debounced flush is auto-cancelled with the test instead of leaking a
|
||||
// pending 30s delay into the test body.
|
||||
val accountant = ResourceUsageAccountant(store, backgroundScope, epochDay = { 1L })
|
||||
val l =
|
||||
RelayUsageListener(
|
||||
accountant = accountant,
|
||||
isMobile = { false },
|
||||
isForeground = { true },
|
||||
nowMs = { now },
|
||||
)
|
||||
block(accountant, l)
|
||||
}
|
||||
|
||||
private suspend fun counters(accountant: ResourceUsageAccountant): Map<String, Long> = accountant.allDaysIncludingLive()[1L].orEmpty()
|
||||
|
||||
@Test
|
||||
fun bucketsASessionByHowLongItLived() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
now = 1_000
|
||||
l.onConnected(r, 10, false)
|
||||
now = 1_000 + 45_000
|
||||
l.onDisconnected(r)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(1L, c[UsageKeys.relayLife(45_000, mobile = false, foreground = true)])
|
||||
assertEquals(1L, c[UsageKeys.relayConnects(mobile = false, foreground = true)])
|
||||
assertEquals(1L, c[UsageKeys.relayDisconnects(mobile = false, foreground = true)])
|
||||
assertNull(c[UsageKeys.RELAY_LIFE_ORPHAN])
|
||||
assertNull(c[UsageKeys.RELAY_LIFE_OVERWRITE])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aDialThatNeverConnectedProducesNoLifetime() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://nos.lol/")
|
||||
l.onConnecting(r)
|
||||
l.onCannotConnect(r, "WebSocket Failure: timeout (SocketTimeoutException)")
|
||||
l.onDisconnected(r)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(1L, c[UsageKeys.relayDials(mobile = false, foreground = true)])
|
||||
assertEquals(1L, c[UsageKeys.relayConnectFails(mobile = false, foreground = true)])
|
||||
// No start stamp to consume: counted as an orphan rather than a 0ms session,
|
||||
// which would otherwise pile into lt5s and fake "instant failures".
|
||||
assertEquals(1L, c[UsageKeys.RELAY_LIFE_ORPHAN])
|
||||
assertNull(c[UsageKeys.relayLife(0, mobile = false, foreground = true)])
|
||||
assertNull(c[UsageKeys.relayConnects(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aSecondConnectBeforeDisconnectIsCountedAsAnOverwrite() =
|
||||
runLedger { accountant, l ->
|
||||
// The teardown race: disconnect(); connect() runs synchronously, so a stale
|
||||
// failure callback for the old socket can land after the new one is open.
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
now = 0
|
||||
l.onConnected(r, 10, false)
|
||||
now = 500_000
|
||||
l.onConnected(r, 10, false)
|
||||
now = 500_100
|
||||
l.onDisconnected(r)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(1L, c[UsageKeys.RELAY_LIFE_OVERWRITE])
|
||||
assertEquals(2L, c[UsageKeys.relayConnects(mobile = false, foreground = true)])
|
||||
// The long session was lost; only the short one is recorded. `connects` is the
|
||||
// denominator that makes that deficit visible instead of silent.
|
||||
assertEquals(1L, c[UsageKeys.relayLife(100, mobile = false, foreground = true)])
|
||||
assertNull(c[UsageKeys.relayLife(500_000, mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun twoRelaysDoNotShareASlot() =
|
||||
runLedger { accountant, l ->
|
||||
val a = relay("wss://relay.damus.io/")
|
||||
val b = relay("wss://nos.lol/")
|
||||
now = 0
|
||||
l.onConnected(a, 10, false)
|
||||
l.onConnected(b, 10, false)
|
||||
now = 90_000
|
||||
l.onDisconnected(a)
|
||||
now = 200_000
|
||||
l.onDisconnected(b)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(1L, c[UsageKeys.relayLife(90_000, mobile = false, foreground = true)])
|
||||
assertEquals(1L, c[UsageKeys.relayLife(200_000, mobile = false, foreground = true)])
|
||||
assertNull(c[UsageKeys.RELAY_LIFE_OVERWRITE])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun sentVerbSplitSumsBackToTheByteTotal() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
val req = ReqCmd("sub1", listOf())
|
||||
val event = EventCmd(mockk(relaxed = true))
|
||||
l.onSent(r, "0123456789", req, success = true)
|
||||
l.onSent(r, "01234", event, success = true)
|
||||
// A failed send is not counted at all, matching relay.msg.*.tx.
|
||||
l.onSent(r, "0123456789012345", req, success = false)
|
||||
|
||||
val c = counters(accountant)
|
||||
val total = c[UsageKeys.relayMsg(mobile = false, foreground = true, received = false)]
|
||||
val split =
|
||||
c.filterKeys { it.startsWith("relay.verb.up.") }.values.sum()
|
||||
assertEquals(15L, total)
|
||||
assertEquals(total, split)
|
||||
assertEquals(10L, c[UsageKeys.relayVerb(ReqCmd.LABEL, received = false, mobile = false, foreground = true)])
|
||||
assertEquals(5L, c[UsageKeys.relayVerb(EventCmd.LABEL, received = false, mobile = false, foreground = true)])
|
||||
// The label is uppercase on the wire; the key segment is not.
|
||||
assertEquals("relay.verb.up.req.wifi.fg", UsageKeys.relayVerb(ReqCmd.LABEL, received = false, mobile = false, foreground = true))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun receivedVerbSplitSumsBackToTheByteTotal() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onIncomingMessage(r, "0123456789", EoseMessage("sub1"))
|
||||
l.onIncomingMessage(r, "012", NoticeMessage("hi"))
|
||||
|
||||
val c = counters(accountant)
|
||||
val total = c[UsageKeys.relayMsg(mobile = false, foreground = true, received = true)]
|
||||
val split = c.filterKeys { it.startsWith("relay.verb.down.") }.values.sum()
|
||||
assertEquals(13L, total)
|
||||
assertEquals(total, split)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun reqsInsideTheConnectWindowCountAsReplay() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
now = 10_000
|
||||
l.onConnected(r, 10, false)
|
||||
// syncState's burst: sent immediately after the socket is ready.
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf()), success = true)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub2", listOf()), success = true)
|
||||
// Well past the window — a user opening a screen, not a replay.
|
||||
now = 10_000 + UsageKeys.REPLAY_WINDOW_MS + 1
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub3", listOf()), success = true)
|
||||
l.onSent(r, "[\"CLOSE\"]", CloseCmd("sub1"), success = true)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(3L, c[UsageKeys.relaySubsSent(mobile = false, foreground = true)])
|
||||
assertEquals(2L, c[UsageKeys.relaySubsReplay(mobile = false, foreground = true)])
|
||||
assertEquals(1L, c[UsageKeys.relaySubsClosed(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aReqOnANeverConnectedRelayIsNotReplay() =
|
||||
runLedger { accountant, l ->
|
||||
// No onConnected, so no start stamp: must not be attributed to a burst.
|
||||
val r = relay("wss://nos.lol/")
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf()), success = true)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(1L, c[UsageKeys.relaySubsSent(mobile = false, foreground = true)])
|
||||
assertNull(c[UsageKeys.relaySubsReplay(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aFailedSendCountsNowhere() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf()), success = false)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertNull(c[UsageKeys.relaySubsSent(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aRefusalNoticeIsCountedByReason() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://nos.lol/")
|
||||
l.onIncomingMessage(r, "[\"NOTICE\",\"x\"]", NoticeMessage("ERROR: too many concurrent REQs"))
|
||||
l.onIncomingMessage(r, "[\"NOTICE\",\"y\"]", NoticeMessage("hello"))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(1L, c[UsageKeys.relayNotice(UsageKeys.NOTICE_TOO_MANY_SUBS)])
|
||||
assertEquals(1L, c[UsageKeys.relayNotice(UsageKeys.NOTICE_UNCLASSIFIED)])
|
||||
// Still part of the byte total, so the verb invariant is unaffected.
|
||||
assertEquals(
|
||||
c[UsageKeys.relayMsg(mobile = false, foreground = true, received = true)],
|
||||
c.filterKeys { it.startsWith("relay.verb.down.") }.values.sum(),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aReqForAnAlreadyOpenSubscriptionCountsAsAResend() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf(Filter())), success = true)
|
||||
// Same subId, still open: the assembler changed its mind.
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf(Filter())), success = true)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub2", listOf(Filter())), success = true)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(3L, c[UsageKeys.relaySubsSent(mobile = false, foreground = true)])
|
||||
assertEquals(1L, c[UsageKeys.relaySubsResent(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aClosedSubscriptionCanBeReopenedWithoutCountingAsAResend() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf(Filter())), success = true)
|
||||
l.onSent(r, "[\"CLOSE\"]", CloseCmd("sub1"), success = true)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf(Filter())), success = true)
|
||||
|
||||
assertNull(counters(accountant)[UsageKeys.relaySubsResent(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aReconnectForgetsOpenSubscriptions() =
|
||||
runLedger { accountant, l ->
|
||||
// The relay forgets them too, so the post-reconnect replay is legitimately
|
||||
// new work and must not be booked as the client changing its mind.
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf(Filter())), success = true)
|
||||
l.onDisconnected(r)
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("sub1", listOf(Filter())), success = true)
|
||||
|
||||
assertNull(counters(accountant)[UsageKeys.relaySubsResent(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun reqsAreAttributedToTheirSubscriptionPurpose() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "0123456789", ReqCmd("a", listOf(ExplainedFilter(purpose = SubPurpose.HOME_FEED))), success = true)
|
||||
l.onSent(r, "01234", ReqCmd("b", listOf(ExplainedFilter(purpose = SubPurpose.OTHER))), success = true)
|
||||
// An assembler #3832 never tagged: its own bucket, not a guess.
|
||||
l.onSent(r, "012", ReqCmd("c", listOf(Filter())), success = true)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(1L, c[UsageKeys.relayPurposeSent("home_feed")])
|
||||
assertEquals(10L, c[UsageKeys.relayPurposeBytes("home_feed")])
|
||||
assertEquals(1L, c[UsageKeys.relayPurposeSent(UsageKeys.PURPOSE_OTHER)])
|
||||
assertEquals(1L, c[UsageKeys.relayPurposeSent(UsageKeys.PURPOSE_UNEXPLAINED)])
|
||||
// Purpose bytes reconcile with the REQ verb total.
|
||||
assertEquals(
|
||||
c[UsageKeys.relayVerb("REQ", received = false, mobile = false, foreground = true)],
|
||||
c.filterKeys { it.startsWith("relay.purpose.") && it.endsWith(".bytes") }.values.sum(),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun handshakeAndDialGapSeparateTheRelayFromOurselves() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
now = 0
|
||||
l.onConnecting(r)
|
||||
// 3s of wall clock to get connected, of which the transport says the
|
||||
// upgrade round trip was 150ms — the other 2850ms is DNS/TCP/TLS/queueing.
|
||||
now = 3_000
|
||||
l.onConnected(r, pingMillis = 150, compressed = false)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(1L, c[UsageKeys.relayHandshake(150, mobile = false, foreground = true)])
|
||||
assertEquals(1L, c[UsageKeys.relayDialGap(2_850, mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun anUntimeableHandshakeRecordsNothingRatherThanZero() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnecting(r)
|
||||
l.onConnected(r, pingMillis = 0, compressed = false)
|
||||
|
||||
val c = counters(accountant)
|
||||
assertNull(c[UsageKeys.relayHandshake(0, mobile = false, foreground = true)])
|
||||
assertNull(c[UsageKeys.relayDialGap(0, mobile = false, foreground = true)])
|
||||
// The connection itself is still counted.
|
||||
assertEquals(1L, c[UsageKeys.relayConnects(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun inboundBytesAreAttributedToTheSubscriptionThatAskedForThem() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("mod1", listOf(ExplainedFilter(purpose = SubPurpose.MODERATION))), success = true)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("feed1", listOf(ExplainedFilter(purpose = SubPurpose.HOME_FEED))), success = true)
|
||||
|
||||
l.onIncomingMessage(r, "0123456789", EventMessage("mod1", mockk(relaxed = true)))
|
||||
l.onIncomingMessage(r, "01234", EventMessage("feed1", mockk(relaxed = true)))
|
||||
l.onIncomingMessage(r, "012", EoseMessage("mod1"))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(13L, c[UsageKeys.relayPurposeDown("moderation")])
|
||||
assertEquals(5L, c[UsageKeys.relayPurposeDown("home_feed")])
|
||||
// Frames, not just bytes: 13 bytes of moderation arrived as two frames, so
|
||||
// the average frame size is recoverable per purpose.
|
||||
assertEquals(2L, c[UsageKeys.relayPurposeDownCount("moderation")])
|
||||
assertEquals(1L, c[UsageKeys.relayPurposeDownCount("home_feed")])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun framesStillInFlightAfterACloseAreStillAttributed() =
|
||||
runLedger { accountant, l ->
|
||||
// The bug this replaced: CLOSE removed the id, so events the relay had
|
||||
// already queued landed in `unattributed`. 8,159 CLOSEs in one session
|
||||
// sent 41% of the download there.
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("s1", listOf(ExplainedFilter(purpose = SubPurpose.HOME_FEED))), success = true)
|
||||
l.onSent(r, "[\"CLOSE\"]", CloseCmd("s1"), success = true)
|
||||
l.onIncomingMessage(r, "0123456789", EventMessage("s1", mockk(relaxed = true)))
|
||||
|
||||
assertEquals(10L, counters(accountant)[UsageKeys.relayPurposeDown("home_feed")])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aFrameArrivingAfterAReconnectIsStillAttributed() =
|
||||
runLedger { accountant, l ->
|
||||
// Purpose is a property of the subscription id, not of the socket, so a
|
||||
// disconnect must not forget it.
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("s1", listOf(ExplainedFilter(purpose = SubPurpose.ENGAGEMENT))), success = true)
|
||||
l.onDisconnected(r)
|
||||
l.onConnected(r, 10, false)
|
||||
l.onIncomingMessage(r, "01234", EventMessage("s1", mockk(relaxed = true)))
|
||||
|
||||
assertEquals(5L, counters(accountant)[UsageKeys.relayPurposeDown("engagement")])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aReconnectStillForgetsWhichSubscriptionsAreOpen() =
|
||||
runLedger { accountant, l ->
|
||||
// The other half of the split: the relay forgot them, so the replay is
|
||||
// new work and must not read as the client changing its mind.
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("s1", listOf(ExplainedFilter(purpose = SubPurpose.HOME_FEED))), success = true)
|
||||
l.onDisconnected(r)
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("s1", listOf(ExplainedFilter(purpose = SubPurpose.HOME_FEED))), success = true)
|
||||
|
||||
assertNull(counters(accountant)[UsageKeys.relaySubsResent(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun anInboundFrameForAnUnknownSubscriptionIsNotGuessedAt() =
|
||||
runLedger { accountant, l ->
|
||||
// Counters wiped mid-session, or a subscription opened before this
|
||||
// connection: attributing it to a purpose would be an invention.
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onIncomingMessage(r, "0123456789", EventMessage("ghost", mockk(relaxed = true)))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(10L, c[UsageKeys.relayPurposeDown(UsageKeys.PURPOSE_UNATTRIBUTED)])
|
||||
assertEquals(1L, c[UsageKeys.relayPurposeDownCount(UsageKeys.PURPOSE_UNATTRIBUTED)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun relayWideFramesAreLeftOutRatherThanMisattributed() =
|
||||
runLedger { accountant, l ->
|
||||
// NOTICE and OK name no subscription, so nothing may be booked for them.
|
||||
// This is why purpose.down deliberately does not reconcile to msg.rx.
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onIncomingMessage(r, "0123456789", NoticeMessage("hello"))
|
||||
l.onIncomingMessage(r, "01234", OkMessage("id", true, ""))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertTrue(c.keys.none { it.startsWith("relay.purpose.") && it.endsWith(".down") })
|
||||
// Still counted in the byte totals and the verb split.
|
||||
assertEquals(15L, c[UsageKeys.relayMsg(mobile = false, foreground = true, received = true)])
|
||||
}
|
||||
|
||||
private fun eventWithId(id: String): EventMessage {
|
||||
val ev = mockk<Event>(relaxed = true)
|
||||
every { ev.id } returns id
|
||||
val msg = mockk<EventMessage>(relaxed = true)
|
||||
every { msg.subId } returns "s1"
|
||||
every { msg.event } returns ev
|
||||
every { msg.label() } returns EventMessage.LABEL
|
||||
return msg
|
||||
}
|
||||
|
||||
@Test
|
||||
fun theSameEventFromTwoRelaysIsCountedOnceAsNewAndOnceAsDuplicate() =
|
||||
runLedger { accountant, l ->
|
||||
// The outbox fan-out asks many relays for the same authors, so one event
|
||||
// arrives once per relay carrying it. That repetition is the measurement.
|
||||
val a = relay("wss://relay.damus.io/")
|
||||
val b = relay("wss://nos.lol/")
|
||||
l.onConnected(a, 10, false)
|
||||
l.onConnected(b, 10, false)
|
||||
val id = "a".repeat(64)
|
||||
l.onIncomingMessage(a, "0123456789", eventWithId(id))
|
||||
l.onIncomingMessage(b, "0123456789", eventWithId(id))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(2L, c[UsageKeys.relayEventsSeen(mobile = false, foreground = true)])
|
||||
assertEquals(1L, c[UsageKeys.relayEventsDup(mobile = false, foreground = true)])
|
||||
assertEquals(10L, c[UsageKeys.relayEventsDupBytes(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun duplicateBytesAreAttributedToThePurposeThatReceivedThem() =
|
||||
runLedger { accountant, l ->
|
||||
// Global duplication says how much is wasted; this says where, which is
|
||||
// what separates "suppress redundant delivery" from "stop fetching it".
|
||||
val a = relay("wss://relay.damus.io/")
|
||||
val b = relay("wss://nos.lol/")
|
||||
l.onConnected(a, 10, false)
|
||||
l.onConnected(b, 10, false)
|
||||
l.onSent(a, "[\"REQ\"]", ReqCmd("s1", listOf(ExplainedFilter(purpose = SubPurpose.USER_PROFILE))), success = true)
|
||||
l.onSent(b, "[\"REQ\"]", ReqCmd("s1", listOf(ExplainedFilter(purpose = SubPurpose.USER_PROFILE))), success = true)
|
||||
|
||||
val id = "b".repeat(64)
|
||||
l.onIncomingMessage(a, "0123456789", eventWithId(id))
|
||||
l.onIncomingMessage(b, "0123456789", eventWithId(id))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(20L, c[UsageKeys.relayPurposeDown("user_profile")])
|
||||
// Only the second copy is waste.
|
||||
assertEquals(10L, c[UsageKeys.relayPurposeDupBytes("user_profile")])
|
||||
assertEquals(10L, c[UsageKeys.relayEventsDupBytes(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aFirstDeliveryIsNeverBookedAsDuplicate() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onSent(r, "[\"REQ\"]", ReqCmd("s1", listOf(ExplainedFilter(purpose = SubPurpose.HOME_FEED))), success = true)
|
||||
l.onIncomingMessage(r, "0123456789", eventWithId("c".repeat(64)))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(10L, c[UsageKeys.relayPurposeDown("home_feed")])
|
||||
assertNull(c[UsageKeys.relayPurposeDupBytes("home_feed")])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun distinctEventsAreNotDuplicates() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
// Differ only past the 64-bit prefix would be a collision; differ within it.
|
||||
l.onIncomingMessage(r, "01234", eventWithId("1".repeat(64)))
|
||||
l.onIncomingMessage(r, "01234", eventWithId("2".repeat(64)))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(2L, c[UsageKeys.relayEventsSeen(mobile = false, foreground = true)])
|
||||
assertNull(c[UsageKeys.relayEventsDup(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aMalformedEventIdIsCountedButNeverDeduplicated() =
|
||||
runLedger { accountant, l ->
|
||||
val r = relay("wss://relay.damus.io/")
|
||||
l.onConnected(r, 10, false)
|
||||
l.onIncomingMessage(r, "01234", eventWithId("short"))
|
||||
l.onIncomingMessage(r, "01234", eventWithId("short"))
|
||||
l.onIncomingMessage(r, "01234", eventWithId("zzzz".repeat(16)))
|
||||
|
||||
val c = counters(accountant)
|
||||
assertEquals(3L, c[UsageKeys.relayEventsSeen(mobile = false, foreground = true)])
|
||||
assertNull(c[UsageKeys.relayEventsDup(mobile = false, foreground = true)])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun everyCommandAndMessageSubtypeHasItsOwnVerb() {
|
||||
// Labels are read off *instances*, because `cmd.label()` is what
|
||||
// RelayUsageListener calls — a subtype whose label() didn't return its own
|
||||
// LABEL would be invisible to a constant-only check. Asserting these against
|
||||
// UsageKeyGrammarTest's lists then keeps the two inventories from drifting,
|
||||
// which they had already started to do.
|
||||
val cmdVerbs =
|
||||
listOf(
|
||||
ReqCmd("s", listOf()),
|
||||
CloseCmd("s"),
|
||||
EventCmd(mockk(relaxed = true)),
|
||||
AuthCmd(mockk(relaxed = true)),
|
||||
CountCmd("s", listOf()),
|
||||
).map { it.label() }
|
||||
val msgVerbs =
|
||||
listOf(
|
||||
EventMessage("s", mockk(relaxed = true)),
|
||||
EoseMessage("s"),
|
||||
OkMessage("id", true, ""),
|
||||
NoticeMessage("m"),
|
||||
AuthMessage("challenge"),
|
||||
ClosedMessage("s", "m"),
|
||||
CountMessage("s", mockk(relaxed = true)),
|
||||
NotifyMessage("m"),
|
||||
LimitsMessage(),
|
||||
).map { it.label() }
|
||||
|
||||
assertEquals(UsageKeyGrammarTest.CMD_LABELS.toSet(), cmdVerbs.toSet())
|
||||
assertEquals(UsageKeyGrammarTest.MSG_LABELS.toSet(), msgVerbs.toSet())
|
||||
|
||||
// No two labels may collide within a direction, and none may be key-hostile
|
||||
// (a dot would inject uncontrolled segments — see UsageKeys.sumMatching).
|
||||
assertEquals(cmdVerbs.size, cmdVerbs.distinct().size)
|
||||
assertEquals(msgVerbs.size, msgVerbs.distinct().size)
|
||||
(cmdVerbs + msgVerbs).forEach {
|
||||
assertFalse("Label '$it' is not usable as a counter key segment", it.isEmpty() || it.contains('.'))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* NOTICE classification. The reason must come from a fixed set: this key is
|
||||
* persisted for 30 days and the text behind it is written by the relay.
|
||||
*/
|
||||
class NoticeReasonTest {
|
||||
@Test
|
||||
fun refusalsAreRecognisedWhateverTheRelayCallsThem() {
|
||||
// strfry's, the one Hypothesis N is about. Its NIP-01 prefix is just
|
||||
// "error", so prefix matching alone would not have caught it.
|
||||
assertEquals(UsageKeys.NOTICE_TOO_MANY_SUBS, UsageKeys.noticeReason("ERROR: too many concurrent REQs"))
|
||||
assertEquals(UsageKeys.NOTICE_TOO_MANY_SUBS, UsageKeys.noticeReason("too many concurrent NEG requests"))
|
||||
assertEquals(UsageKeys.NOTICE_TOO_MANY_SUBS, UsageKeys.noticeReason("blocked: too many subscriptions"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun standardPrefixesAreCategorised() {
|
||||
assertEquals(UsageKeys.NOTICE_AUTH_REQUIRED, UsageKeys.noticeReason("auth-required: we need to know you"))
|
||||
assertEquals(UsageKeys.NOTICE_RATE_LIMITED, UsageKeys.noticeReason("rate-limited: slow down"))
|
||||
assertEquals(UsageKeys.NOTICE_RESTRICTED, UsageKeys.noticeReason("restricted: not on the allowlist"))
|
||||
assertEquals(UsageKeys.NOTICE_INVALID, UsageKeys.noticeReason("invalid: bad filter"))
|
||||
assertEquals(UsageKeys.NOTICE_BLOCKED, UsageKeys.noticeReason("blocked: you are banned"))
|
||||
assertEquals(UsageKeys.NOTICE_ERROR, UsageKeys.noticeReason("error: something broke"))
|
||||
}
|
||||
|
||||
/** Verbatim from a device on 2026-08-02 — the wordings the allowlist was missing. */
|
||||
@Test
|
||||
fun realWorldNoticesAreClassified() {
|
||||
// Refusals. Each one drops a REQ that then never EOSEs, so its `since` never
|
||||
// advances and syncState re-sends it on every reconnect.
|
||||
assertEquals(UsageKeys.NOTICE_QUERY_COST, UsageKeys.noticeReason("Kgo0HH: closed: too many steps"))
|
||||
assertEquals(UsageKeys.NOTICE_QUERY_COST, UsageKeys.noticeReason("too many kinds"))
|
||||
assertEquals(UsageKeys.NOTICE_REQ_REFUSED, UsageKeys.noticeReason("Denied! This relay does not accept REQs."))
|
||||
|
||||
// Chatter that costs bytes and means nothing.
|
||||
assertEquals(UsageKeys.NOTICE_BENIGN, UsageKeys.noticeReason("keepalive"))
|
||||
assertEquals(UsageKeys.NOTICE_BENIGN, UsageKeys.noticeReason("as7rp4: PERF: [/!\\ LS] 1087 scan, 0 dedup, 500 match"))
|
||||
|
||||
// A bare subscription id carries no meaning and must stay unclassified rather
|
||||
// than being read as a machine-readable prefix.
|
||||
assertEquals(UsageKeys.NOTICE_UNCLASSIFIED, UsageKeys.noticeReason("AccountFollowsLoaderSubAssemblerxE1r8A"))
|
||||
assertEquals(UsageKeys.NOTICE_UNCLASSIFIED, UsageKeys.noticeReason("Kgo0HH"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aSubscriptionIdPrefixDoesNotHideTheRealOne() {
|
||||
// These relays send "<subId>: <prefix>: <text>", which makes the subId the
|
||||
// prefix and buries the standard one behind it.
|
||||
assertEquals(UsageKeys.NOTICE_AUTH_REQUIRED, UsageKeys.noticeReason("sub123: auth-required: need auth"))
|
||||
assertEquals(UsageKeys.NOTICE_RATE_LIMITED, UsageKeys.noticeReason("sub123: rate-limited: slow down"))
|
||||
// Still works without the prefix.
|
||||
assertEquals(UsageKeys.NOTICE_AUTH_REQUIRED, UsageKeys.noticeReason("auth-required: need auth"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun freeFormProseNeverBecomesAKey() {
|
||||
// The bug RelayObserver had to fix: without a fixed output set, cardinality
|
||||
// grows with the number of distinct sentences relays happen to write.
|
||||
listOf(
|
||||
"hello there",
|
||||
"Please contact admin@example.com for access",
|
||||
"",
|
||||
"::::",
|
||||
"a".repeat(5000),
|
||||
).forEach {
|
||||
val reason = UsageKeys.noticeReason(it)
|
||||
assertTrue("'$it' produced unlisted reason '$reason'", reason in UsageKeys.NOTICE_REASONS)
|
||||
}
|
||||
assertEquals(UsageKeys.NOTICE_UNCLASSIFIED, UsageKeys.noticeReason("hello there"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun anUnlistedReasonCannotMintAKey() {
|
||||
assertEquals(UsageKeys.relayNotice(UsageKeys.NOTICE_UNCLASSIFIED), UsageKeys.relayNotice("something-invented"))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The report is sent as a NIP-17 DM, so its size is a correctness property, not a
|
||||
* cosmetic one: relays cap events and an oversized report is simply never delivered.
|
||||
*/
|
||||
class ReportSizeTest {
|
||||
private fun day(keys: Int) = (0 until keys).associate { "relay.purpose.p$it.bytes" to 123_456L }
|
||||
|
||||
@Test
|
||||
fun aFullLedgerStaysSendable() {
|
||||
// 30 retained days at the density a busy day now produces.
|
||||
val days = (1L..30L).associateWith { day(1_200) }
|
||||
val report = ResourceUsageReportAssembler().buildReport(days, today = 30L)
|
||||
|
||||
assertTrue("report was ${report.length} chars", report.length < 100_000)
|
||||
assertTrue("nothing was said about the omission", report.contains("omitted to keep this report sendable"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aSmallLedgerIsNotTruncatedAndSaysNothingAboutOmission() {
|
||||
val days = (1L..3L).associateWith { day(20) }
|
||||
val report = ResourceUsageReportAssembler().buildReport(days, today = 3L)
|
||||
|
||||
assertTrue(report.contains("day 1 "))
|
||||
assertTrue(report.contains("day 3 "))
|
||||
assertFalse(report.contains("omitted"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun theNewestDayIsKeptEvenIfItAloneExceedsTheBudget() {
|
||||
// Dropping everything would leave a report that says nothing at all.
|
||||
val days = mapOf(1L to day(50), 2L to day(20_000))
|
||||
val report = ResourceUsageReportAssembler().buildReport(days, today = 2L)
|
||||
|
||||
assertTrue("newest day missing", report.contains("day 2 "))
|
||||
assertTrue(report.contains("1 earlier day(s) omitted"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun omittedDaysStillCountTowardTheSummaryTables() {
|
||||
// The tables are built from every day; only the raw dump is bounded.
|
||||
val days = (1L..30L).associateWith { mapOf(UsageKeys.relayConnects(mobile = false, foreground = true) to 10L) }
|
||||
val report = ResourceUsageReportAssembler().buildReport(days, today = 30L)
|
||||
|
||||
// 7 days x 10 connections in the week table.
|
||||
assertTrue(report.contains("| Relay reconnections | 70 "))
|
||||
}
|
||||
}
|
||||
|
||||
+5
-1
@@ -206,7 +206,11 @@ open class BasicRelayClient(
|
||||
!msg.startsWith("failed to connect to /127.0.0.1") &&
|
||||
msg != "Socket closed" &&
|
||||
msg != "Socket is closed" &&
|
||||
msg != "Cancelled"
|
||||
// OkHttp spells it with one L (RealCall throws
|
||||
// IOException("Canceled")). "Cancelled" never matched, so
|
||||
// client-initiated cancels were being reported as connection
|
||||
// failures and inflating the relay.connfails counter.
|
||||
msg != "Canceled"
|
||||
)
|
||||
) {
|
||||
if (code != null || response != null) {
|
||||
|
||||
Reference in New Issue
Block a user