diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt index 0c59012d38..0c3936c413 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt @@ -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 diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/RelayProxyClientConnector.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/RelayProxyClientConnector.kt index 995870ce21..b7283b37d2 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/RelayProxyClientConnector.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/RelayProxyClientConnector.kt @@ -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, 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 diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/RelayUsageListener.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/RelayUsageListener.kt index 5d97ef56cb..79e4e9e9c6 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/RelayUsageListener.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/RelayUsageListener.kt @@ -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() + + /** + * 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>() + + /** + * 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() + + /** + * 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() + + /** Relay -> when this dial was decided, so the pre-request cost can be separated from the handshake. */ + private val dialStartedAt = ConcurrentHashMap() + + /** Notice texts already logged, so one wording costs one line however often it arrives. */ + private val loggedNotices = ConcurrentHashMap.newKeySet() + 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 + } } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageAccountant.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageAccountant.kt index eaa90e8d10..e7807eb02b 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageAccountant.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageAccountant.kt @@ -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..*`), 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 { diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageReportAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageReportAssembler.kt index 73c4f510f3..9b289e5ee7 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageReportAssembler.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageReportAssembler.kt @@ -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)) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageStore.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageStore.kt index 7e7824631b..c25bd7e3f5 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageStore.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageStore.kt @@ -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. */ diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/UsageKeys.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/UsageKeys.kt index e43cde73a9..75271e3b7f 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/UsageKeys.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/resourceusage/UsageKeys.kt @@ -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 `.[.]` 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 = 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>() + private val VERB_DOWN_KEYS = ConcurrentHashMap>() + + /** + * `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.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 + } } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/AgentConsoleViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/AgentConsoleViewModel.kt index 98d19c7937..2433459063 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/AgentConsoleViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/AgentConsoleViewModel.kt @@ -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() } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzDmListViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzDmListViewModel.kt index a619471e3f..dccf0cb851 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzDmListViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzDmListViewModel.kt @@ -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 diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzJoinReconnect.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzJoinReconnect.kt new file mode 100644 index 0000000000..9d419d535b --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzJoinReconnect.kt @@ -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) +} diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzRelayImportViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzRelayImportViewModel.kt index 9cded7263f..a3976d203f 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzRelayImportViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/buzz/BuzzRelayImportViewModel.kt @@ -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. diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageLedgerTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageLedgerTest.kt index ac4c066788..c42b3bdada 100644 --- a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageLedgerTest.kt +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/resourceusage/ResourceUsageLedgerTest.kt @@ -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 { + val out = mutableMapOf() + 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 = + 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(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 = 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(relaxed = true) + every { ev.id } returns id + val msg = mockk(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 ": : ", 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")) + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt index d0f1b7bc16..eec65748e9 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt @@ -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) {