From 96371715f4d8ea077ce267aaf167c05b1d857db3 Mon Sep 17 00:00:00 2001 From: nrobi144 Date: Thu, 18 Jun 2026 09:49:25 +0300 Subject: [PATCH] feat(commons): per-relay latency tracker + slow-relay classifier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of the desktop relay-latency-health feature: add a rolling-window latency tracker that decorates the quartz RelayConnectionListener, plus a pure classifier that flags relays whose per-metric p50 exceeds 2× the cohort median. No store integration or UI yet — those come in follow-up commits. commons commonMain (CLI-safe, no Compose runtime, no JVM-only deps): - LatencyMetric: OK_ACK / EOSE / FIRST_RESULT / PING - MetricSample: @Immutable (p50Ms, count) - RelayLatencySnapshot: @Immutable, backed by ImmutableMap so strong skipping engages when unchanged rows are re-emitted - SlowReason: @Immutable (metric, relayP50, cohortP50, multiplier) - HealthReason sealed interface: Unresponsive(gap) | Slow(SlowReason) - classifySlowRelays(): pure. Honors NIP-11 auth_required / payment_required (paid/auth-only relays are excluded from both cohort and target until auth completes — otherwise they'd be perpetually flagged while CLOSED'ing anonymous queries). commons jvmAndroid (ConcurrentHashMap is JVM-only): - LatencyRingBuffer: fixed-capacity (default 50) IntArray ring, synchronized push, snapshotMedian / snapshotSamples / restore. - RelayLatencyTracker: pending-eventId / pending-subId / firstResultSeen maps + per-(relay, metric) ring buffers. Handles every pairing rule the deepened plan called out: * onSent EventCmd → record eventId timestamp * onSent ReqCmd → record subId timestamp; clear firstResultSeen * onSent CloseCmd → drop pending subId (no sample) — prevents ComposeSubscriptionManager's sub-id reuse from pairing late events with a new REQ * success=false → no-op (websocket buffer was full) * OkMessage → pair by eventId, push OK_ACK * EventMessage → first-only, push FIRST_RESULT * EoseMessage → pair by subId, push EOSE * ClosedMessage → drop pending (fast negative response, not a latency signal — was previously recording 300s TTL samples for any auth-required relay) * onConnected → push PING * onDisconnected → drop all pending (no TTL samples) * sweep(now) → TTL-expire pending entries (60s OK / 300s REQ), record TTL value as the sample AUTH retries: the second onSent overwrites the timestamp, so samples reflect the retry leg — matches the user's mental model of "speed of the actual publish". Pending maps are size-capped at 256 entries per relay as a safety net against adversarial relays. Tracker owns no CoroutineScope — RelayHealthStore drives sweep + snapshot from its existing 60s reclassify tick (Phase 2). - RelayLatencyListener: thin RelayConnectionListener decorator, installInto / uninstallFrom paralleling RelayHealthListener. Tests: - LatencyRingBufferTest (7): wrap, median odd/even, restore from larger or smaller arrays, chronological snapshotSamples. - RelayLatencyTrackerTest (13): OK pairing, EOSE + FIRST_RESULT pairing, success=false ignore, CloseCmd drops pending, ClosedMessage drops pending, AUTH retry overwrites timestamp, disconnect drops all, sweep TTL semantics (OK vs REQ), FIRST_RESULT only sampled when not yet seen, ping, per-relay isolation, 256-entry cap, restore round-trip. - ClassifySlowRelaysTest (9): empty, Tor short-circuit, cohort < 2, 2× flag, count-below-min excludes from cohort, NIP-11 auth_required excludes / includes once auth complete, payment_required excludes, worst-metric-multiplier wins when multiple flag, exact-2× does not flag (strict greater-than). Note: a pre-existing RelayHealthStoreCloseTest case on the base branch (fix/relay-health-threading-and-sleep-resume) hangs in advanceUntilIdle. Not related to this commit; new tests pass cleanly with a tighter test filter. Will revisit when integrating Phase 2. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../relays/health/ClassifySlowRelays.kt | 120 ++++++++ .../commons/relays/health/HealthReason.kt | 43 +++ .../commons/relays/health/LatencyMetric.kt | 38 +++ .../commons/relays/health/MetricSample.kt | 30 ++ .../relays/health/RelayLatencySnapshot.kt | 43 +++ .../commons/relays/health/SlowReason.kt | 38 +++ .../relays/health/ClassifySlowRelaysTest.kt | 208 +++++++++++++ .../relays/health/LatencyRingBuffer.kt | 109 +++++++ .../relays/health/RelayLatencyListener.kt | 75 +++++ .../relays/health/RelayLatencyTracker.kt | 290 ++++++++++++++++++ .../relays/health/LatencyRingBufferTest.kt | 91 ++++++ .../relays/health/RelayLatencyTrackerTest.kt | 239 +++++++++++++++ 12 files changed, 1324 insertions(+) create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/ClassifySlowRelays.kt create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/HealthReason.kt create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyMetric.kt create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/MetricSample.kt create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencySnapshot.kt create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/SlowReason.kt create mode 100644 commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/ClassifySlowRelaysTest.kt create mode 100644 commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyRingBuffer.kt create mode 100644 commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyListener.kt create mode 100644 commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyTracker.kt create mode 100644 commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyRingBufferTest.kt create mode 100644 commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyTrackerTest.kt diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/ClassifySlowRelays.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/ClassifySlowRelays.kt new file mode 100644 index 0000000000..90a6d9db38 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/ClassifySlowRelays.kt @@ -0,0 +1,120 @@ +/* + * 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.commons.relays.health + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation +import kotlinx.collections.immutable.ImmutableMap +import kotlinx.collections.immutable.persistentMapOf +import kotlinx.collections.immutable.toPersistentMap + +const val DEFAULT_SLOW_MIN_SAMPLES: Int = 5 +const val DEFAULT_SLOW_MULTIPLIER: Double = 2.0 +const val DEFAULT_SLOW_MIN_COHORT_SIZE: Int = 2 + +/** + * Pure classifier. Returns the relays whose median latency is more than [multiplier] × the + * cohort median on at least one metric. Per-metric cohort = relays with ≥ [minSamples] samples + * for that metric. + * + * Gates (return early with empty map): + * - Tor mode is on (existing gate; relay timing through Tor is lossy on purpose). + * - Relays whose NIP-11 [Nip11RelayInformation.limitation] advertises `auth_required` or + * `payment_required` AND [authStatus] reports auth incomplete are excluded from both + * cohort *and* targets. Otherwise paid / auth-only relays would be perpetually flagged + * while sending mostly CLOSED responses to anonymous queries. + * + * If a relay is slow on multiple metrics, the metric with the worst multiplier wins. + */ +fun classifySlowRelays( + snapshots: ImmutableMap, + nip11: ImmutableMap, + torEnabled: Boolean = false, + minSamples: Int = DEFAULT_SLOW_MIN_SAMPLES, + multiplier: Double = DEFAULT_SLOW_MULTIPLIER, + minCohortSize: Int = DEFAULT_SLOW_MIN_COHORT_SIZE, + authStatus: (NormalizedRelayUrl) -> Boolean, +): ImmutableMap { + if (torEnabled) return persistentMapOf() + if (snapshots.isEmpty()) return persistentMapOf() + + val eligible: Map = + snapshots.filter { (url, _) -> includeInClassification(url, nip11, authStatus) } + if (eligible.size < minCohortSize) return persistentMapOf() + + val worst = mutableMapOf() + + for (metric in LatencyMetric.entries) { + // Per-metric cohort = relays whose own count ≥ minSamples for this metric. + val cohort: List> = + eligible.mapNotNull { (url, snap) -> + val sample = snap.samples[metric] ?: return@mapNotNull null + if (sample.count < minSamples) return@mapNotNull null + url to sample.p50Ms + } + if (cohort.size < minCohortSize) continue + + val cohortP50 = medianOf(cohort.map { it.second }) + if (cohortP50 <= 0) continue + + for ((url, relayP50) in cohort) { + val mult = relayP50.toDouble() / cohortP50.toDouble() + if (mult > multiplier) { + val existing = worst[url] + if (existing == null || mult > existing.multiplier) { + worst[url] = + SlowReason( + metric = metric, + relayP50Ms = relayP50, + cohortP50Ms = cohortP50, + multiplier = mult, + ) + } + } + } + } + + return worst.toPersistentMap() +} + +private fun includeInClassification( + url: NormalizedRelayUrl, + nip11: ImmutableMap, + authStatus: (NormalizedRelayUrl) -> Boolean, +): Boolean { + val info = nip11[url] ?: return true + val limit = info.limitation ?: return true + val requiresAuth = limit.auth_required == true || limit.payment_required == true + if (!requiresAuth) return true + return authStatus(url) +} + +/** + * Median of an Int list (n=1..N is fine). For even N returns the lower of the two middles + * (`values[n / 2 - 1]`) — avoids floating-point arithmetic and keeps results stable. + * Caller is responsible for non-empty input. + */ +internal fun medianOf(values: List): Int { + if (values.isEmpty()) return 0 + val sorted = values.sorted() + val n = sorted.size + return if (n % 2 == 1) sorted[n / 2] else sorted[n / 2 - 1] +} diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/HealthReason.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/HealthReason.kt new file mode 100644 index 0000000000..635c07a871 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/HealthReason.kt @@ -0,0 +1,43 @@ +/* + * 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.commons.relays.health + +import androidx.compose.runtime.Immutable + +/** + * Reason a relay is flagged in the unhealthy/slow surfaces. Drives the per-row chip text + * in [com.vitorpamplona.amethyst.commons.relays.health.ui.UnhealthyRelayRow] and the + * dashboard slow-chip. + */ +@Immutable +sealed interface HealthReason { + /** Last activity was more than `gapSeconds` ago (existing 7-day classifier). */ + @Immutable + data class Unresponsive( + val gapSeconds: Long, + ) : HealthReason + + /** Median latency exceeded the cohort threshold on at least one metric. */ + @Immutable + data class Slow( + val slow: SlowReason, + ) : HealthReason +} diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyMetric.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyMetric.kt new file mode 100644 index 0000000000..d56dceefb7 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyMetric.kt @@ -0,0 +1,38 @@ +/* + * 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.commons.relays.health + +/** + * Per-relay latency signals that feed the slow-relay classifier. + * + * - [OK_ACK]: time from EVENT sent → relay's OK message for that event_id (NIP-01). + * - [EOSE]: time from REQ sent → relay's EOSE for that sub_id. + * - [FIRST_RESULT]: time from REQ sent → first matching EVENT for that sub_id. Filter-dependent; + * labelled "First result" in the UI with a tooltip caveat — see the deepened plan. + * - [PING]: per-(re)connect TCP/WS handshake time captured by [com.vitorpamplona.quartz + * .nip01Core.relay.client.listeners.RelayConnectionListener.onConnected]. + */ +enum class LatencyMetric { + OK_ACK, + EOSE, + FIRST_RESULT, + PING, +} diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/MetricSample.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/MetricSample.kt new file mode 100644 index 0000000000..9ef33feee4 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/MetricSample.kt @@ -0,0 +1,30 @@ +/* + * 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.commons.relays.health + +import androidx.compose.runtime.Immutable + +/** Aggregate of one metric's rolling samples on one relay. `count == 0` means no samples yet. */ +@Immutable +data class MetricSample( + val p50Ms: Int, + val count: Int, +) diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencySnapshot.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencySnapshot.kt new file mode 100644 index 0000000000..32b769a913 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencySnapshot.kt @@ -0,0 +1,43 @@ +/* + * 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.commons.relays.health + +import androidx.compose.runtime.Immutable +import kotlinx.collections.immutable.ImmutableMap +import kotlinx.collections.immutable.persistentMapOf + +/** + * Snapshot of one relay's rolling-window latency at a point in time. Compose-safe — backed by + * an [ImmutableMap] so strong skipping engages when nothing changed since the previous tick. + * + * Missing metrics simply aren't in the map (no entry vs zero-count is semantically the same; + * the classifier treats both as "no samples"). + */ +@Immutable +data class RelayLatencySnapshot( + val samples: ImmutableMap = persistentMapOf(), +) { + /** Convenience: returns the count for [metric], or 0 if no samples exist. */ + fun countOf(metric: LatencyMetric): Int = samples[metric]?.count ?: 0 + + /** Convenience: returns the p50 for [metric], or null if no samples exist. */ + fun p50Of(metric: LatencyMetric): Int? = samples[metric]?.p50Ms +} diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/SlowReason.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/SlowReason.kt new file mode 100644 index 0000000000..79e09724e4 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/SlowReason.kt @@ -0,0 +1,38 @@ +/* + * 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.commons.relays.health + +import androidx.compose.runtime.Immutable + +/** + * The "why" of a slow flag: which metric tipped over and by how much. Surfaced on the + * dashboard chip ("Slow: OK 2.4×") and in the unhealthy-relays popup. + * + * If a relay is slow on multiple metrics, the classifier picks the metric with the worst + * multiplier for display — the rest are visible in the NIP-11 popup's per-metric breakdown. + */ +@Immutable +data class SlowReason( + val metric: LatencyMetric, + val relayP50Ms: Int, + val cohortP50Ms: Int, + val multiplier: Double, +) diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/ClassifySlowRelaysTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/ClassifySlowRelaysTest.kt new file mode 100644 index 0000000000..4d293005ff --- /dev/null +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/ClassifySlowRelaysTest.kt @@ -0,0 +1,208 @@ +/* + * 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.commons.relays.health + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation +import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation.RelayInformationLimitation +import kotlinx.collections.immutable.persistentMapOf +import kotlinx.collections.immutable.toPersistentMap +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class ClassifySlowRelaysTest { + private val fast1 = NormalizedRelayUrl("wss://fast1.test/") + private val fast2 = NormalizedRelayUrl("wss://fast2.test/") + private val fast3 = NormalizedRelayUrl("wss://fast3.test/") + private val slow = NormalizedRelayUrl("wss://slow.test/") + private val paid = NormalizedRelayUrl("wss://paid.test/") + + private fun snap(vararg pairs: Pair): RelayLatencySnapshot = RelayLatencySnapshot(samples = mapOf(*pairs).toPersistentMap()) + + private fun authAlwaysComplete( + @Suppress("UNUSED_PARAMETER") url: NormalizedRelayUrl, + ): Boolean = true + + @Test + fun emptySnapshotsReturnsEmpty() { + val result = + classifySlowRelays( + snapshots = persistentMapOf(), + nip11 = persistentMapOf(), + authStatus = ::authAlwaysComplete, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun torEnabledShortCircuitsToEmpty() { + val result = + classifySlowRelays( + snapshots = persistentMapOf(slow to snap(LatencyMetric.OK_ACK to MetricSample(10_000, 10))), + nip11 = persistentMapOf(), + authStatus = ::authAlwaysComplete, + torEnabled = true, + ) + assertTrue(result.isEmpty()) + } + + @Test + fun belowMinCohortSizeReturnsEmpty() { + // Only one relay has samples — no cohort to compare against. + val snapshots = + persistentMapOf( + slow to snap(LatencyMetric.OK_ACK to MetricSample(5_000, 50)), + ) + val result = classifySlowRelays(snapshots, persistentMapOf(), authStatus = ::authAlwaysComplete) + assertTrue(result.isEmpty()) + } + + @Test + fun flagsRelayWith2xCohortMedian() { + val snapshots = + persistentMapOf( + fast1 to snap(LatencyMetric.OK_ACK to MetricSample(200, 10)), + fast2 to snap(LatencyMetric.OK_ACK to MetricSample(250, 10)), + fast3 to snap(LatencyMetric.OK_ACK to MetricSample(300, 10)), + slow to snap(LatencyMetric.OK_ACK to MetricSample(2_000, 10)), + ) + val result = classifySlowRelays(snapshots, persistentMapOf(), authStatus = ::authAlwaysComplete) + val reason = result[slow] + assertNotNull(reason) + assertEquals(LatencyMetric.OK_ACK, reason.metric) + assertEquals(2_000, reason.relayP50Ms) + // Cohort medians of [200, 250, 300, 2000] sorted = lower middle = 250. + assertEquals(250, reason.cohortP50Ms) + assertEquals(8.0, reason.multiplier) + assertNull(result[fast1]) + assertNull(result[fast2]) + assertNull(result[fast3]) + } + + @Test + fun belowMinSamplesExcludesRelayFromCohort() { + val snapshots = + persistentMapOf( + fast1 to snap(LatencyMetric.OK_ACK to MetricSample(200, 4)), // below min + fast2 to snap(LatencyMetric.OK_ACK to MetricSample(250, 10)), + slow to snap(LatencyMetric.OK_ACK to MetricSample(2_000, 10)), + ) + val result = classifySlowRelays(snapshots, persistentMapOf(), authStatus = ::authAlwaysComplete) + // Cohort is now [fast2, slow] — slow vs cohort median 250 → 8× → flagged. + assertEquals(8.0, result[slow]!!.multiplier) + } + + @Test + fun nip11AuthRequiredExcludesRelayFromCohortAndTargets() { + val snapshots = + persistentMapOf( + fast1 to snap(LatencyMetric.OK_ACK to MetricSample(200, 10)), + fast2 to snap(LatencyMetric.OK_ACK to MetricSample(250, 10)), + slow to snap(LatencyMetric.OK_ACK to MetricSample(2_000, 10)), + paid to snap(LatencyMetric.OK_ACK to MetricSample(50_000, 10)), + ) + val nip11 = + persistentMapOf( + paid to Nip11RelayInformation(limitation = RelayInformationLimitation(auth_required = true)), + ) + // `paid` hasn't completed auth, so it is excluded from both cohort and targets. + val result = classifySlowRelays(snapshots, nip11) { url -> url != paid } + // `paid` is excluded (auth_required + auth not complete). `slow` still flagged. + assertNotNull(result[slow]) + assertNull(result[paid]) + } + + @Test + fun nip11AuthRequiredIncludesRelayOnceAuthComplete() { + val snapshots = + persistentMapOf( + fast1 to snap(LatencyMetric.OK_ACK to MetricSample(200, 10)), + fast2 to snap(LatencyMetric.OK_ACK to MetricSample(250, 10)), + paid to snap(LatencyMetric.OK_ACK to MetricSample(50_000, 10)), + ) + val nip11 = + persistentMapOf( + paid to Nip11RelayInformation(limitation = RelayInformationLimitation(auth_required = true)), + ) + // Auth complete everywhere → `paid` participates as cohort and target. + val result = classifySlowRelays(snapshots, nip11) { _ -> true } + assertNotNull(result[paid]) + assertEquals(LatencyMetric.OK_ACK, result[paid]!!.metric) + } + + @Test + fun paymentRequiredExcludesLikeAuthRequired() { + val snapshots = + persistentMapOf( + fast1 to snap(LatencyMetric.OK_ACK to MetricSample(200, 10)), + fast2 to snap(LatencyMetric.OK_ACK to MetricSample(250, 10)), + paid to snap(LatencyMetric.OK_ACK to MetricSample(50_000, 10)), + ) + val nip11 = + persistentMapOf( + paid to Nip11RelayInformation(limitation = RelayInformationLimitation(payment_required = true)), + ) + val result = classifySlowRelays(snapshots, nip11) { url -> url != paid } + assertNull(result[paid]) + } + + @Test + fun worstMetricMultiplierWinsWhenMultipleFlag() { + val snapshots = + persistentMapOf( + fast1 to + snap( + LatencyMetric.OK_ACK to MetricSample(200, 10), + LatencyMetric.EOSE to MetricSample(300, 10), + ), + fast2 to + snap( + LatencyMetric.OK_ACK to MetricSample(250, 10), + LatencyMetric.EOSE to MetricSample(350, 10), + ), + slow to + snap( + LatencyMetric.OK_ACK to MetricSample(1_000, 10), // 4× + LatencyMetric.EOSE to MetricSample(4_000, 10), // 11.4× (cohort p50 = 350) + ), + ) + val result = classifySlowRelays(snapshots, persistentMapOf(), authStatus = ::authAlwaysComplete) + val reason = result[slow]!! + assertEquals(LatencyMetric.EOSE, reason.metric) + assertTrue(reason.multiplier > 10.0) + } + + @Test + fun rightOnThresholdDoesNotFlag() { + // exactly 2× the cohort median — multiplier > 2.0 required to flag. + val snapshots = + persistentMapOf( + fast1 to snap(LatencyMetric.OK_ACK to MetricSample(100, 10)), + fast2 to snap(LatencyMetric.OK_ACK to MetricSample(100, 10)), + slow to snap(LatencyMetric.OK_ACK to MetricSample(200, 10)), + ) + val result = classifySlowRelays(snapshots, persistentMapOf(), authStatus = ::authAlwaysComplete) + assertNull(result[slow]) + } +} diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyRingBuffer.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyRingBuffer.kt new file mode 100644 index 0000000000..82dc4f83f6 --- /dev/null +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyRingBuffer.kt @@ -0,0 +1,109 @@ +/* + * 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.commons.relays.health + +/** + * Fixed-capacity ring of `Int` millisecond samples. New pushes overwrite the oldest sample + * once full. Thread-safe via `synchronized(this)` — push is called from relay-network threads, + * [snapshotMedian] and [snapshotSamples] are called from the [RelayHealthStore] classifier + * coroutine on the 60 s tick. + * + * Median is computed by copying the live region into a fresh array and sorting — the buffer + * itself is not reordered. For [DEFAULT_CAPACITY] = 50 the per-snapshot sort is sub-millisecond + * and the snapshot happens only on the classifier tick, so allocation cost is irrelevant. + */ +class LatencyRingBuffer( + val capacity: Int = DEFAULT_CAPACITY, +) { + init { + require(capacity > 0) { "capacity must be > 0; got $capacity" } + } + + private val data = IntArray(capacity) + private var nextIndex = 0 + private var filled = 0 + + /** Number of valid samples currently in the buffer (0..capacity). */ + val size: Int + @Synchronized get() = filled + + /** Append a sample. If the buffer is full the oldest sample is overwritten. */ + @Synchronized + fun push(ms: Int) { + data[nextIndex] = ms + nextIndex = (nextIndex + 1) % capacity + if (filled < capacity) filled++ + } + + /** Median of the current samples in ms, or `null` if empty. */ + @Synchronized + fun snapshotMedian(): Int? { + if (filled == 0) return null + val copy = IntArray(filled) + if (filled < capacity) { + // Buffer not yet full: valid samples are at [0, filled). + System.arraycopy(data, 0, copy, 0, filled) + } else { + // Buffer full: nextIndex points at oldest, wrap-copy in order. + val tailLen = capacity - nextIndex + System.arraycopy(data, nextIndex, copy, 0, tailLen) + if (nextIndex > 0) System.arraycopy(data, 0, copy, tailLen, nextIndex) + } + copy.sort() + val n = copy.size + return if (n % 2 == 1) copy[n / 2] else copy[n / 2 - 1] + } + + /** + * Returns the live samples as a fresh `IntArray` (chronological order from oldest to + * newest). Used by the persistence layer to encode the buffer. + */ + @Synchronized + fun snapshotSamples(): IntArray { + if (filled == 0) return EMPTY + val out = IntArray(filled) + if (filled < capacity) { + System.arraycopy(data, 0, out, 0, filled) + } else { + val tailLen = capacity - nextIndex + System.arraycopy(data, nextIndex, out, 0, tailLen) + if (nextIndex > 0) System.arraycopy(data, 0, out, tailLen, nextIndex) + } + return out + } + + /** + * Replaces the current contents with [samples] (used to restore from persisted state). + * Only the most recent [capacity] entries of [samples] are kept. + */ + @Synchronized + fun restore(samples: IntArray) { + val src = if (samples.size > capacity) samples.copyOfRange(samples.size - capacity, samples.size) else samples + for (i in src.indices) data[i] = src[i] + filled = src.size + nextIndex = src.size % capacity + } + + companion object { + const val DEFAULT_CAPACITY: Int = 50 + private val EMPTY = IntArray(0) + } +} diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyListener.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyListener.kt new file mode 100644 index 0000000000..0b9a29b672 --- /dev/null +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyListener.kt @@ -0,0 +1,75 @@ +/* + * 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.commons.relays.health + +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +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.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command + +/** + * Decorates the quartz [RelayConnectionListener] to feed events into a [RelayLatencyTracker]. + * Install once per process via [installInto] alongside [RelayHealthListener]. + * + * All forwarded methods are non-suspending and safe from any thread; the tracker handles its + * own concurrency. + */ +class RelayLatencyListener( + private val tracker: RelayLatencyTracker, +) : RelayConnectionListener { + override fun onConnected( + relay: IRelayClient, + pingMillis: Int, + compressed: Boolean, + ) { + tracker.recordPing(relay.url, pingMillis) + } + + override fun onSent( + relay: IRelayClient, + cmdStr: String, + cmd: Command, + success: Boolean, + ) { + tracker.recordSent(relay.url, cmd, success) + } + + override fun onIncomingMessage( + relay: IRelayClient, + msgStr: String, + msg: Message, + ) { + tracker.recordIncoming(relay.url, msg) + } + + override fun onDisconnected(relay: IRelayClient) { + tracker.recordDisconnect(relay.url) + } + + fun installInto(client: INostrClient) { + client.addConnectionListener(this) + } + + fun uninstallFrom(client: INostrClient) { + client.removeConnectionListener(this) + } +} diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyTracker.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyTracker.kt new file mode 100644 index 0000000000..07e3714416 --- /dev/null +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyTracker.kt @@ -0,0 +1,290 @@ +/* + * 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.commons.relays.health + +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage +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.OkMessage +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.EventCmd +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import kotlinx.collections.immutable.ImmutableMap +import kotlinx.collections.immutable.persistentMapOf +import kotlinx.collections.immutable.toPersistentMap +import java.util.concurrent.ConcurrentHashMap + +/** + * Per-relay rolling-window latency tracker. + * + * **Pairing rules** (matches the plan's table): + * + * | Trigger | Action | + * |---------|--------| + * | `onSent(EventCmd, success=true)` | pending.eventId[relay][id] = now | + * | `onSent(ReqCmd, success=true)` | pending.subId[relay][subId] = now; clear firstResultSeen | + * | `onSent(CloseCmd, success=true)` | drop pending.subId[relay][subId] (no sample) | + * | `onSent(_, success=false)` | ignore (websocket buffer was full / socket closing) | + * | `OkMessage` | match by eventId, push delta into OK_ACK | + * | `EventMessage` | if firstResultSeen[relay][subId] is clear, push delta into FIRST_RESULT | + * | `EoseMessage` | match by subId, push delta into EOSE | + * | `ClosedMessage` | drop pending.subId (no sample — fast negative response) | + * | `onConnected` | push pingMillis into PING ring | + * | `sweep(now)` | TTL-expire pending entries (60 s OK / 300 s REQ) | + * | `onDisconnected`| drop ALL pending for that relay (no TTL samples recorded) | + * + * **AUTH retries**: the second `onSent` for the same (relay, eventId) overwrites the timestamp, + * so OK_ACK samples reflect the retry leg (not the AUTH round-trip). Intentional — matches the + * user's mental model of "how fast was the actual publish". + * + * **Threading**: All public methods are non-suspending and safe to call from any thread. + * Pending maps are [ConcurrentHashMap]s; samples are guarded by per-buffer `synchronized`. + * + * **Lifecycle**: This tracker owns no [kotlinx.coroutines.CoroutineScope]. The + * [RelayHealthStore] is responsible for calling [sweep] periodically and reading [snapshot] + * on its existing 60-second classifier tick. + */ +class RelayLatencyTracker( + val ringCapacity: Int = LatencyRingBuffer.DEFAULT_CAPACITY, + val maxPendingPerRelay: Int = DEFAULT_MAX_PENDING_PER_RELAY, + val okTtlMs: Long = DEFAULT_OK_TTL_MS, + val reqTtlMs: Long = DEFAULT_REQ_TTL_MS, +) { + // Per-relay pending maps. `Long` is `currentTimeMillis()` at the time of the send. + private val pendingEventId = ConcurrentHashMap>() + private val pendingSubId = ConcurrentHashMap>() + + // Tracks which sub-ids have already produced a FIRST_RESULT sample so subsequent + // EVENT messages for the same sub-id short-circuit cheaply (no map lookup beyond + // the per-relay set). + private val firstResultSeen = ConcurrentHashMap>() + + // samples[relay][metric] -> ring buffer of latency in ms. + private val samples = + ConcurrentHashMap>() + + // ------ Recording API ------ + + /** + * Called from `RelayConnectionListener.onSent`. Only records when [success] is true (a + * `false` means the websocket send buffer rejected the message — it never went out). + */ + fun recordSent( + relay: NormalizedRelayUrl, + cmd: Command, + success: Boolean, + nowMs: Long = System.currentTimeMillis(), + ) { + if (!success) return + when (cmd) { + is EventCmd -> putPending(pendingEventId, relay, cmd.event.id, nowMs) + is ReqCmd -> { + putPending(pendingSubId, relay, cmd.subId, nowMs) + firstResultSeen[relay]?.remove(cmd.subId) + } + is CloseCmd -> { + pendingSubId[relay]?.remove(cmd.subId) + firstResultSeen[relay]?.remove(cmd.subId) + } + else -> Unit + } + } + + /** Called from `RelayConnectionListener.onIncomingMessage`. */ + fun recordIncoming( + relay: NormalizedRelayUrl, + msg: Message, + nowMs: Long = System.currentTimeMillis(), + ) { + when (msg) { + is OkMessage -> { + val sentAt = pendingEventId[relay]?.remove(msg.eventId) ?: return + ringFor(relay, LatencyMetric.OK_ACK).push((nowMs - sentAt).toInt().coerceAtLeast(0)) + } + is EoseMessage -> { + val sentAt = pendingSubId[relay]?.remove(msg.subId) ?: return + ringFor(relay, LatencyMetric.EOSE).push((nowMs - sentAt).toInt().coerceAtLeast(0)) + firstResultSeen[relay]?.remove(msg.subId) + } + is EventMessage -> { + val seenSet = firstResultSeen.computeIfAbsent(relay) { newConcurrentSet() } + if (msg.subId in seenSet) return + val sentAt = pendingSubId[relay]?.get(msg.subId) ?: return + seenSet.add(msg.subId) + ringFor(relay, LatencyMetric.FIRST_RESULT) + .push((nowMs - sentAt).toInt().coerceAtLeast(0)) + } + is ClosedMessage -> { + pendingSubId[relay]?.remove(msg.subId) + firstResultSeen[relay]?.remove(msg.subId) + } + else -> Unit + } + } + + /** Called from `RelayConnectionListener.onConnected`. */ + fun recordPing( + relay: NormalizedRelayUrl, + pingMs: Int, + ) { + if (pingMs < 0) return + ringFor(relay, LatencyMetric.PING).push(pingMs) + } + + /** Called from `RelayConnectionListener.onDisconnected`. Drops all pending entries. */ + fun recordDisconnect(relay: NormalizedRelayUrl) { + pendingEventId[relay]?.clear() + pendingSubId[relay]?.clear() + firstResultSeen[relay]?.clear() + } + + // ------ TTL sweep ------ + + /** + * Expires pending entries older than the configured TTLs and records the TTL value as the + * sample (per the brainstorm: "punish silent relays"). Idempotent and cheap. + */ + fun sweep(nowMs: Long = System.currentTimeMillis()) { + for ((relay, pending) in pendingEventId) { + val it = pending.entries.iterator() + while (it.hasNext()) { + val (_, sentAt) = it.next() + if (nowMs - sentAt >= okTtlMs) { + ringFor(relay, LatencyMetric.OK_ACK).push(okTtlMs.toInt()) + it.remove() + } + } + } + for ((relay, pending) in pendingSubId) { + val it = pending.entries.iterator() + while (it.hasNext()) { + val entry = it.next() + val sentAt = entry.value + if (nowMs - sentAt >= reqTtlMs) { + ringFor(relay, LatencyMetric.EOSE).push(reqTtlMs.toInt()) + // Record a FIRST_RESULT TTL sample only if the relay never emitted any + // matching event for this sub-id. + val seenSet = firstResultSeen[relay] + if (seenSet == null || entry.key !in seenSet) { + ringFor(relay, LatencyMetric.FIRST_RESULT).push(reqTtlMs.toInt()) + } + seenSet?.remove(entry.key) + it.remove() + } + } + } + } + + // ------ Snapshot ------ + + /** Immutable snapshot of all tracked relays' current rolling-window medians. */ + fun snapshot(): ImmutableMap { + if (samples.isEmpty()) return persistentMapOf() + val out = HashMap(samples.size) + for ((relay, perMetric) in samples) { + val builder = HashMap(perMetric.size) + for ((metric, ring) in perMetric) { + val median = ring.snapshotMedian() ?: continue + builder[metric] = MetricSample(p50Ms = median, count = ring.size) + } + if (builder.isNotEmpty()) { + out[relay] = RelayLatencySnapshot(builder.toPersistentMap()) + } + } + return out.toPersistentMap() + } + + /** + * Raw per-relay per-metric sample arrays (chronological order). Used by the persistence + * layer to encode the rings into the packed `lat_` Preferences key. + */ + fun samplesForPersistence(): Map> { + if (samples.isEmpty()) return emptyMap() + val out = HashMap>(samples.size) + for ((relay, perMetric) in samples) { + val inner = HashMap(perMetric.size) + for ((metric, ring) in perMetric) { + val arr = ring.snapshotSamples() + if (arr.isNotEmpty()) inner[metric] = arr + } + if (inner.isNotEmpty()) out[relay] = inner + } + return out + } + + /** + * Restore samples (typically from the persistence layer at app start). Overwrites any + * existing buffers for the listed `(relay, metric)` pairs; other relays/metrics are + * untouched. + */ + fun restoreSamples(saved: Map>) { + for ((relay, perMetric) in saved) { + for ((metric, arr) in perMetric) { + if (arr.isEmpty()) continue + ringFor(relay, metric).restore(arr) + } + } + } + + // ------ Internals ------ + + private fun ringFor( + relay: NormalizedRelayUrl, + metric: LatencyMetric, + ): LatencyRingBuffer = + samples + .computeIfAbsent(relay) { ConcurrentHashMap() } + .computeIfAbsent(metric) { LatencyRingBuffer(ringCapacity) } + + private fun putPending( + store: ConcurrentHashMap>, + relay: NormalizedRelayUrl, + key: String, + nowMs: Long, + ) { + val perRelay = store.computeIfAbsent(relay) { synchronizedLinkedMap() } + synchronized(perRelay) { + // Cap protects against adversarial relays that open thousands of sub-ids — TTL + // sweep is the primary expiry mechanism; this is a safety net. + if (perRelay.size >= maxPendingPerRelay && key !in perRelay) { + val first = perRelay.entries.iterator() + if (first.hasNext()) { + first.next() + first.remove() + } + } + perRelay[key] = nowMs + } + } + + private fun synchronizedLinkedMap(): MutableMap = java.util.Collections.synchronizedMap(java.util.LinkedHashMap(16, 0.75f, false)) + + private fun newConcurrentSet(): MutableSet = java.util.Collections.newSetFromMap(ConcurrentHashMap()) + + companion object { + const val DEFAULT_MAX_PENDING_PER_RELAY: Int = 256 + const val DEFAULT_OK_TTL_MS: Long = 60_000L + const val DEFAULT_REQ_TTL_MS: Long = 300_000L + } +} diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyRingBufferTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyRingBufferTest.kt new file mode 100644 index 0000000000..35dcf93094 --- /dev/null +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/LatencyRingBufferTest.kt @@ -0,0 +1,91 @@ +/* + * 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.commons.relays.health + +import org.junit.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertNull + +class LatencyRingBufferTest { + @Test + fun emptyBufferReturnsNullMedianAndZeroSize() { + val r = LatencyRingBuffer(5) + assertNull(r.snapshotMedian()) + assertEquals(0, r.size) + assertEquals(0, r.snapshotSamples().size) + } + + @Test + fun medianOddCountIsMiddleSample() { + val r = LatencyRingBuffer(5) + listOf(100, 300, 200).forEach { r.push(it) } + assertEquals(200, r.snapshotMedian()) + assertEquals(3, r.size) + } + + @Test + fun medianEvenCountReturnsLowerMiddle() { + val r = LatencyRingBuffer(4) + listOf(100, 200, 300, 400).forEach { r.push(it) } + // Lower middle is values.sorted()[n/2 - 1] = 200 + assertEquals(200, r.snapshotMedian()) + } + + @Test + fun pushBeyondCapacityOverwritesOldest() { + val r = LatencyRingBuffer(3) + listOf(1000, 1000, 1000, 50, 50, 50).forEach { r.push(it) } + // Only the last 3 (all 50) remain. + assertEquals(3, r.size) + assertEquals(50, r.snapshotMedian()) + } + + @Test + fun snapshotSamplesReturnsChronologicalOrder() { + val r = LatencyRingBuffer(4) + listOf(10, 20, 30).forEach { r.push(it) } + assertContentEquals(intArrayOf(10, 20, 30), r.snapshotSamples()) + // After wrap, oldest-to-newest is 20, 30, 40, 50 + r.push(40) + r.push(50) + assertContentEquals(intArrayOf(20, 30, 40, 50), r.snapshotSamples()) + } + + @Test + fun restoreFromArrayWhenSmallerThanCapacity() { + val r = LatencyRingBuffer(5) + r.restore(intArrayOf(100, 200, 300)) + assertEquals(3, r.size) + assertContentEquals(intArrayOf(100, 200, 300), r.snapshotSamples()) + // Subsequent pushes append at the tail. + r.push(400) + assertContentEquals(intArrayOf(100, 200, 300, 400), r.snapshotSamples()) + } + + @Test + fun restoreFromArrayLargerThanCapacityKeepsMostRecent() { + val r = LatencyRingBuffer(3) + r.restore(intArrayOf(100, 200, 300, 400, 500)) + assertEquals(3, r.size) + assertContentEquals(intArrayOf(300, 400, 500), r.snapshotSamples()) + } +} diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyTrackerTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyTrackerTest.kt new file mode 100644 index 0000000000..6f7703a129 --- /dev/null +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayLatencyTrackerTest.kt @@ -0,0 +1,239 @@ +/* + * 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.commons.relays.health + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage +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.OkMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd +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 org.junit.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class RelayLatencyTrackerTest { + private val relay = NormalizedRelayUrl("wss://relay.test/") + private val relayB = NormalizedRelayUrl("wss://other.test/") + + private fun fakeEvent(idHex: String): Event = + Event( + id = idHex.padEnd(64, '0'), + pubKey = "pub".padEnd(64, '0'), + createdAt = 0L, + kind = 1, + tags = emptyArray(), + content = "", + sig = "sig".padEnd(128, '0'), + ) + + private fun req(subId: String) = ReqCmd(subId, listOf(Filter())) + + @Test + fun okAckPairsByEventIdAndPushesDelta() { + val tracker = RelayLatencyTracker() + val event = fakeEvent("a") + tracker.recordSent(relay, EventCmd(event), success = true, nowMs = 1_000) + tracker.recordIncoming(relay, OkMessage(event.id, true, ""), nowMs = 1_250) + + val snap = tracker.snapshot()[relay] + assertNotNull(snap) + val ok = snap.samples[LatencyMetric.OK_ACK] + assertNotNull(ok) + assertEquals(250, ok.p50Ms) + assertEquals(1, ok.count) + } + + @Test + fun eoseAndFirstResultPairBySubId() { + val tracker = RelayLatencyTracker() + tracker.recordSent(relay, req("s1"), success = true, nowMs = 1_000) + tracker.recordIncoming(relay, EventMessage("s1", fakeEvent("e")), nowMs = 1_100) + tracker.recordIncoming(relay, EventMessage("s1", fakeEvent("f")), nowMs = 1_400) // ignored + tracker.recordIncoming(relay, EoseMessage("s1"), nowMs = 1_500) + + val snap = tracker.snapshot()[relay]!! + assertEquals(100, snap.samples[LatencyMetric.FIRST_RESULT]!!.p50Ms) + assertEquals(1, snap.samples[LatencyMetric.FIRST_RESULT]!!.count) + assertEquals(500, snap.samples[LatencyMetric.EOSE]!!.p50Ms) + assertEquals(1, snap.samples[LatencyMetric.EOSE]!!.count) + } + + @Test + fun unsuccessfulSendDoesNotInsertPending() { + val tracker = RelayLatencyTracker() + val event = fakeEvent("a") + tracker.recordSent(relay, EventCmd(event), success = false, nowMs = 1_000) + tracker.recordIncoming(relay, OkMessage(event.id, true, ""), nowMs = 1_250) + assertTrue(tracker.snapshot().isEmpty()) + } + + @Test + fun closeCmdDropsPendingReq() { + val tracker = RelayLatencyTracker() + tracker.recordSent(relay, req("s1"), success = true, nowMs = 1_000) + tracker.recordSent(relay, CloseCmd("s1"), success = true, nowMs = 1_100) + // Now a stale EOSE for s1 arrives — should be ignored. + tracker.recordIncoming(relay, EoseMessage("s1"), nowMs = 1_500) + assertTrue(tracker.snapshot().isEmpty()) + } + + @Test + fun closedMessageDropsPendingReqAndDoesNotSample() { + val tracker = RelayLatencyTracker() + tracker.recordSent(relay, req("s1"), success = true, nowMs = 1_000) + tracker.recordIncoming(relay, ClosedMessage("s1", "auth-required"), nowMs = 1_020) + // Subsequent stale EOSE is a no-op. + tracker.recordIncoming(relay, EoseMessage("s1"), nowMs = 1_500) + assertTrue(tracker.snapshot().isEmpty()) + } + + @Test + fun authRetryOverwritesPendingTimestamp() { + val tracker = RelayLatencyTracker() + val event = fakeEvent("a") + tracker.recordSent(relay, EventCmd(event), success = true, nowMs = 1_000) + // AUTH happens, then quartz re-sends the same EventCmd. + tracker.recordSent(relay, EventCmd(event), success = true, nowMs = 6_000) + tracker.recordIncoming(relay, OkMessage(event.id, true, ""), nowMs = 6_200) + + val snap = tracker.snapshot()[relay]!! + val ok = snap.samples[LatencyMetric.OK_ACK]!! + // Sample reflects the retry leg (200 ms), not the AUTH round-trip (5.2 s). + assertEquals(200, ok.p50Ms) + } + + @Test + fun disconnectDropsAllPendingForRelay() { + val tracker = RelayLatencyTracker() + val event = fakeEvent("a") + tracker.recordSent(relay, EventCmd(event), success = true, nowMs = 1_000) + tracker.recordSent(relay, req("s1"), success = true, nowMs = 1_000) + tracker.recordDisconnect(relay) + // Late OK / EOSE no longer pair. + tracker.recordIncoming(relay, OkMessage(event.id, true, ""), nowMs = 1_200) + tracker.recordIncoming(relay, EoseMessage("s1"), nowMs = 1_300) + assertTrue(tracker.snapshot().isEmpty()) + } + + @Test + fun sweepRecordsTtlValueAsSample() { + val tracker = RelayLatencyTracker(okTtlMs = 60_000L, reqTtlMs = 300_000L) + val event = fakeEvent("a") + tracker.recordSent(relay, EventCmd(event), success = true, nowMs = 0) + tracker.recordSent(relay, req("s1"), success = true, nowMs = 0) + + tracker.sweep(nowMs = 60_001L) + + val snap = tracker.snapshot()[relay]!! + // OK_ACK: TTL expired → recorded as 60_000 ms. + assertEquals(60_000, snap.samples[LatencyMetric.OK_ACK]!!.p50Ms) + // EOSE: REQ TTL is 5min, not yet expired → no sample. + assertNull(snap.samples[LatencyMetric.EOSE]) + + tracker.sweep(nowMs = 300_001L) + val snap2 = tracker.snapshot()[relay]!! + // EOSE + FIRST_RESULT TTL recorded. + assertEquals(300_000, snap2.samples[LatencyMetric.EOSE]!!.p50Ms) + assertEquals(300_000, snap2.samples[LatencyMetric.FIRST_RESULT]!!.p50Ms) + } + + @Test + fun firstResultTtlOnlyRecordedWhenNothingWasSeen() { + val tracker = RelayLatencyTracker(reqTtlMs = 300_000L) + tracker.recordSent(relay, req("s1"), success = true, nowMs = 0) + tracker.recordIncoming(relay, EventMessage("s1", fakeEvent("e")), nowMs = 100) + // FIRST_RESULT was already sampled. EOSE never arrived. Sweep should record only an + // EOSE TTL sample, not a FIRST_RESULT TTL sample. + tracker.sweep(nowMs = 300_001L) + + val snap = tracker.snapshot()[relay]!! + assertEquals(100, snap.samples[LatencyMetric.FIRST_RESULT]!!.p50Ms) + assertEquals(300_000, snap.samples[LatencyMetric.EOSE]!!.p50Ms) + } + + @Test + fun pingRecordsDirectly() { + val tracker = RelayLatencyTracker() + tracker.recordPing(relay, 42) + tracker.recordPing(relay, 58) + val snap = tracker.snapshot()[relay]!! + assertEquals(42, snap.samples[LatencyMetric.PING]!!.p50Ms) // lower-middle of 2 + assertEquals(2, snap.samples[LatencyMetric.PING]!!.count) + } + + @Test + fun perRelayIsolation() { + val tracker = RelayLatencyTracker() + val event = fakeEvent("a") + tracker.recordSent(relay, EventCmd(event), success = true, nowMs = 1_000) + // Same event_id arriving on a *different* relay — should not pair. + tracker.recordIncoming(relayB, OkMessage(event.id, true, ""), nowMs = 1_200) + assertTrue(tracker.snapshot().isEmpty()) + } + + @Test + fun pendingMapSizeCappedAtMaxPerRelay() { + val tracker = RelayLatencyTracker(maxPendingPerRelay = 4) + repeat(10) { i -> + tracker.recordSent(relay, req("s$i"), success = true, nowMs = i.toLong()) + } + // Sweep with low enough now — none of these expired. Yet the pending set is capped. + // We can't directly observe internal state, but a stale-sub EOSE for the *first* + // sub-id should now miss (oldest dropped). + tracker.recordIncoming(relay, EoseMessage("s0"), nowMs = 100) + assertTrue(tracker.snapshot().isEmpty()) + + // A still-pending later sub-id pairs correctly. + tracker.recordIncoming(relay, EoseMessage("s9"), nowMs = 200) + val snap = tracker.snapshot()[relay] + assertNotNull(snap) + } + + @Test + fun restoreSamplesRoundTripsThroughPersistenceShape() { + val tracker = RelayLatencyTracker() + tracker.restoreSamples( + mapOf( + relay to + mapOf( + LatencyMetric.OK_ACK to intArrayOf(100, 200, 300), + LatencyMetric.EOSE to intArrayOf(50, 60), + ), + ), + ) + val snap = tracker.snapshot()[relay]!! + assertEquals(200, snap.samples[LatencyMetric.OK_ACK]!!.p50Ms) + assertEquals(3, snap.samples[LatencyMetric.OK_ACK]!!.count) + assertEquals(50, snap.samples[LatencyMetric.EOSE]!!.p50Ms) + assertEquals(2, snap.samples[LatencyMetric.EOSE]!!.count) + + // Persistence shape symmetric. + val back = tracker.samplesForPersistence()[relay]!! + assertEquals(2, back.size) + } +}