mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-10-06 03:38:23 +00:00
feat(resourceusage): relay churn and traffic attribution counters
The ledger could say how much relay data the app moved, but not why. It
counted completed connections and a single undifferentiated byte total,
so "1.65 GB/day across 6,600 connects" could not be broken down further,
and relay.connfails was being read as a dial-failure count when it also
fires for mid-session drops of successful connections.
Adds, all as counters with no behaviour change:
relay.dials / relay.disc real dial and disconnect counts
relay.life.<bucket> connection-lifetime histogram, bucketed to
straddle STABLE_CONNECTION_IN_SECS
relay.verb.up/down.<verb> the byte totals split by protocol verb
relay.purpose.<p>.* REQ bytes, inbound bytes and frames by the
SubPurpose that asked, read off the
ExplainedFilter that already travels on the
filter
relay.subs.* REQs sent, closed, replayed after connect,
and re-sent for an already-open subscription
relay.events.* inbound EVENT frames and how many carried an
event already delivered
relay.notice.<reason> NOTICE frames by an allowlisted reason
relay.hs / relay.gap the transport's own handshake timing, and
everything before the request went out
relay.trigger.<cause> which decision asked for a reconnect
This commit is contained in:
davotoula
@@ -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
|
||||
|
||||
+262
-3
@@ -20,10 +20,22 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.service.resourceusage
|
||||
|
||||
import android.os.SystemClock
|
||||
import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.ExplainedFilter
|
||||
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,12 +43,89 @@ 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: at the window size below that is
|
||||
* the difference between ~200 KB and several MB on a 512 MB-class device, for a
|
||||
* counter that only has to spot repetition. 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. */
|
||||
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,
|
||||
@@ -44,7 +133,44 @@ class RelayUsageListener(
|
||||
success: Boolean,
|
||||
) {
|
||||
if (success) {
|
||||
accountant.add(UsageKeys.relayMsg(isMobile(), isForeground(), received = false), cmdStr.length.toLong())
|
||||
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.label()) {
|
||||
ReqCmd.LABEL -> {
|
||||
accountant.add(UsageKeys.relaySubsSent(mobile, fg), 1)
|
||||
|
||||
val purpose = purposeOf(cmd)
|
||||
accountant.add(UsageKeys.relayPurposeSent(purpose), 1)
|
||||
accountant.add(UsageKeys.relayPurposeBytes(purpose), bytes)
|
||||
|
||||
// Already open on this connection, so this REQ replaces a live
|
||||
// subscription rather than starting one.
|
||||
val subId = (cmd as ReqCmd).subId
|
||||
if (subPurpose.size >= MAX_TRACKED_SUBS) subPurpose.clear()
|
||||
subPurpose[subId] = purpose
|
||||
|
||||
val known = openSubs.getOrPut(relay.url) { ConcurrentHashMap.newKeySet() }
|
||||
if (!known.add(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)
|
||||
}
|
||||
}
|
||||
CloseCmd.LABEL -> {
|
||||
accountant.add(UsageKeys.relaySubsClosed(mobile, fg), 1)
|
||||
openSubs[relay.url]?.remove((cmd as CloseCmd).subId)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -53,7 +179,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 +246,90 @@ 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)
|
||||
// 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)
|
||||
dialStartedAt.remove(relay.url)?.let { startedAt ->
|
||||
val gap = nowMs() - startedAt - 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: Command): String {
|
||||
val filters = (cmd as? ReqCmd)?.filters ?: return UsageKeys.PURPOSE_UNEXPLAINED
|
||||
val explained =
|
||||
filters.firstOrNull { it is ExplainedFilter } as? ExplainedFilter
|
||||
?: return UsageKeys.PURPOSE_UNEXPLAINED
|
||||
return UsageKeys.purposeKeyPart(explained.purpose)
|
||||
}
|
||||
|
||||
/** The first 64 bits of an event id, or null if it is not a well-formed id. */
|
||||
private fun idPrefix(id: String): Long? = if (id.length < 16) null else runCatching { id.substring(0, 16).toULong(16).toLong() }.getOrNull()
|
||||
|
||||
/** 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)
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
+13
-3
@@ -37,8 +37,15 @@ import java.util.concurrent.atomic.AtomicLong
|
||||
* AtomicLong (not LongAdder) because draining must be loss-free: getAndSet(0)
|
||||
* 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.
|
||||
* in the map after a drain — the key space is small and *mostly* fixed
|
||||
* (dims x areas), so this costs a few hundred boxed zeros at most.
|
||||
*
|
||||
* The exception is the per-relay churn counters (`relay.host.<host>.*`), whose
|
||||
* cardinality follows the user's relay list rather than a compile-time set:
|
||||
* two keys per relay, so a few hundred more entries for a large list. Still
|
||||
* negligible in memory, but it does mean neither this map nor the persisted
|
||||
* store has a fixed upper bound any more. Keep that in mind before adding
|
||||
* another counter keyed on runtime data.
|
||||
*
|
||||
* 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,7 +81,10 @@ 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
|
||||
if (flushScheduled.compareAndSet(false, true)) {
|
||||
scope.launch {
|
||||
|
||||
+5
-2
@@ -29,8 +29,9 @@ import java.util.Locale
|
||||
* Assembles the Markdown resource-usage report the user can DM to the
|
||||
* developers via NIP-17 — same shape as the crash ReportAssembler: a device
|
||||
* header table, a human-readable summary, then the full per-day counter dump
|
||||
* as the technical payload. Counters are sizes/durations/counts only; no
|
||||
* URLs, relay names, or content.
|
||||
* as the technical payload. Counters are sizes/durations/counts only, and never
|
||||
* content.
|
||||
*
|
||||
*/
|
||||
class ResourceUsageReportAssembler {
|
||||
fun buildReport(
|
||||
@@ -132,6 +133,8 @@ class ResourceUsageReportAssembler {
|
||||
/** Markdown table header/body separator row. */
|
||||
private const val TABLE_SEPARATOR = "| --- | --- |\n"
|
||||
|
||||
/** Caps how many relay hosts a shared report can name. See the class doc. */
|
||||
|
||||
fun formatBytes(bytes: Long): String =
|
||||
when {
|
||||
bytes >= 1024L * 1024L * 1024L -> String.format(Locale.US, "%.2f GB", bytes / (1024.0 * 1024.0 * 1024.0))
|
||||
|
||||
+5
-1
@@ -34,7 +34,11 @@ 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, plus
|
||||
* two keys per relay per day now that the churn counters are keyed on runtime
|
||||
* data (see [ResourceUsageAccountant], which owns that caveat). The whole file is
|
||||
* re-serialized on every flush (debounced to ~30s while traffic flows), so that
|
||||
* growth 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.
|
||||
*/
|
||||
|
||||
+593
-17
@@ -20,6 +20,11 @@
|
||||
*/
|
||||
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.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 +34,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 +79,39 @@ 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> = Array(DIMS.size) { if (suffix == null) "$prefix.${DIMS[it]}" else "$prefix.${DIMS[it]}.$suffix" }
|
||||
|
||||
/** `net.image.mobile.bg.rx` — HTTP bytes for a subsystem. */
|
||||
fun net(
|
||||
role: String,
|
||||
@@ -87,25 +145,526 @@ 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
|
||||
"rate-limited" in prefixes || ("rate" in text && "limit" in text) -> NOTICE_RATE_LIMITED
|
||||
"auth-required" in prefixes || ("auth" in text && "required" in text) -> NOTICE_AUTH_REQUIRED
|
||||
"restricted" in prefixes -> NOTICE_RESTRICTED
|
||||
"invalid" in prefixes -> NOTICE_INVALID
|
||||
"blocked" in prefixes -> NOTICE_BLOCKED
|
||||
"unsupported" in prefixes -> NOTICE_UNSUPPORTED
|
||||
"error" 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
|
||||
|
||||
/**
|
||||
* Half-open upper bounds, in ms, for the [relayLife] histogram. Deliberately
|
||||
* straddles [SHORT_SESSION_MS]: a mean cannot tell a tight cluster sitting on
|
||||
* that bar from a bimodal mix; this can.
|
||||
*/
|
||||
private val LIFE_BUCKET_BOUNDS_MS =
|
||||
longArrayOf(5_000, 30_000, SHORT_SESSION_MS, 120_000, 300_000).also {
|
||||
// [lifeBucketIndex] linear-scans for the first bound greater than the
|
||||
// elapsed time, so the bounds must ascend. One of them 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(it.asList() == it.sorted()) {
|
||||
"relay.life bounds must ascend, got ${it.toList()}. " +
|
||||
"SHORT_SESSION_MS is ${SHORT_SESSION_MS}ms — reorder the bounds to match."
|
||||
}
|
||||
}
|
||||
|
||||
/** Derived from the bounds so a bound change can never leave a label lying about it. */
|
||||
private val LIFE_BUCKET_NAMES =
|
||||
Array(LIFE_BUCKET_BOUNDS_MS.size + 1) { i ->
|
||||
if (i < LIFE_BUCKET_BOUNDS_MS.size) {
|
||||
"lt${LIFE_BUCKET_BOUNDS_MS[i] / 1000}s"
|
||||
} else {
|
||||
"gte${LIFE_BUCKET_BOUNDS_MS.last() / 1000}s"
|
||||
}
|
||||
}
|
||||
|
||||
private fun lifeBucketIndex(elapsedMs: Long): Int {
|
||||
for (i in LIFE_BUCKET_BOUNDS_MS.indices) {
|
||||
if (elapsedMs < LIFE_BUCKET_BOUNDS_MS[i]) return i
|
||||
}
|
||||
return LIFE_BUCKET_BOUNDS_MS.size
|
||||
}
|
||||
|
||||
fun lifeBucket(elapsedMs: Long): String = LIFE_BUCKET_NAMES[lifeBucketIndex(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 = RELAY_LIFE[lifeBucketIndex(elapsedMs)][dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_LIFE = Array(LIFE_BUCKET_NAMES.size) { dimKeys("relay.life.${LIFE_BUCKET_NAMES[it]}") }
|
||||
|
||||
/**
|
||||
* `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.
|
||||
*/
|
||||
fun relayPurposeSent(purpose: String): String = "relay.purpose.$purpose.sent"
|
||||
|
||||
fun relayPurposeBytes(purpose: String): String = "relay.purpose.$purpose.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 = "relay.purpose.$purpose.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 = "relay.purpose.$purpose.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 = "relay.purpose.$purpose.dupbytes"
|
||||
|
||||
/** 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.
|
||||
*/
|
||||
fun purposeKeyPart(purpose: SubPurpose): String = 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, for the two connect-timing histograms below.
|
||||
*/
|
||||
private val CONNECT_BUCKET_BOUNDS_MS = longArrayOf(100, 500, 2_000, 10_000, 30_000)
|
||||
private val CONNECT_BUCKET_NAMES =
|
||||
Array(CONNECT_BUCKET_BOUNDS_MS.size + 1) { i ->
|
||||
if (i < CONNECT_BUCKET_BOUNDS_MS.size) "lt${CONNECT_BUCKET_BOUNDS_MS[i]}ms" else "gte${CONNECT_BUCKET_BOUNDS_MS.last()}ms"
|
||||
}
|
||||
|
||||
private fun connectBucketIndex(ms: Long): Int {
|
||||
for (i in CONNECT_BUCKET_BOUNDS_MS.indices) {
|
||||
if (ms < CONNECT_BUCKET_BOUNDS_MS[i]) return i
|
||||
}
|
||||
return CONNECT_BUCKET_BOUNDS_MS.size
|
||||
}
|
||||
|
||||
fun connectBucket(ms: Long): String = CONNECT_BUCKET_NAMES[connectBucketIndex(ms)]
|
||||
|
||||
/**
|
||||
* `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 = RELAY_HS[connectBucketIndex(ms)][dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_HS = Array(CONNECT_BUCKET_NAMES.size) { dimKeys("relay.hs.${CONNECT_BUCKET_NAMES[it]}") }
|
||||
|
||||
/**
|
||||
* `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 = RELAY_GAP[connectBucketIndex(ms)][dimIndex(mobile, foreground)]
|
||||
|
||||
private val RELAY_GAP = Array(CONNECT_BUCKET_NAMES.size) { dimKeys("relay.gap.${CONNECT_BUCKET_NAMES[it]}") }
|
||||
|
||||
/**
|
||||
* `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 +746,35 @@ 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
|
||||
outer@ for ((key, value) in this) {
|
||||
for (part in parts) {
|
||||
if (!key.hasSegment(part)) continue@outer
|
||||
}
|
||||
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.
|
||||
|
||||
+774
@@ -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,756 @@ 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"))
|
||||
}
|
||||
}
|
||||
|
||||
+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