mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-08 23:54:39 +00:00
feat(commons): per-relay latency tracker + slow-relay classifier
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) <noreply@anthropic.com>
This commit is contained in:
nrobi144
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
24655a8a89
commit
96371715f4
+120
@@ -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<NormalizedRelayUrl, RelayLatencySnapshot>,
|
||||
nip11: ImmutableMap<NormalizedRelayUrl, Nip11RelayInformation?>,
|
||||
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<NormalizedRelayUrl, SlowReason> {
|
||||
if (torEnabled) return persistentMapOf()
|
||||
if (snapshots.isEmpty()) return persistentMapOf()
|
||||
|
||||
val eligible: Map<NormalizedRelayUrl, RelayLatencySnapshot> =
|
||||
snapshots.filter { (url, _) -> includeInClassification(url, nip11, authStatus) }
|
||||
if (eligible.size < minCohortSize) return persistentMapOf()
|
||||
|
||||
val worst = mutableMapOf<NormalizedRelayUrl, SlowReason>()
|
||||
|
||||
for (metric in LatencyMetric.entries) {
|
||||
// Per-metric cohort = relays whose own count ≥ minSamples for this metric.
|
||||
val cohort: List<Pair<NormalizedRelayUrl, Int>> =
|
||||
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<NormalizedRelayUrl, Nip11RelayInformation?>,
|
||||
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>): 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]
|
||||
}
|
||||
+43
@@ -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
|
||||
}
|
||||
+38
@@ -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,
|
||||
}
|
||||
+30
@@ -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,
|
||||
)
|
||||
+43
@@ -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<LatencyMetric, MetricSample> = 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
|
||||
}
|
||||
+38
@@ -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,
|
||||
)
|
||||
+208
@@ -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<LatencyMetric, MetricSample>): 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])
|
||||
}
|
||||
}
|
||||
+109
@@ -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)
|
||||
}
|
||||
}
|
||||
+75
@@ -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)
|
||||
}
|
||||
}
|
||||
+290
@@ -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<NormalizedRelayUrl, MutableMap<String, Long>>()
|
||||
private val pendingSubId = ConcurrentHashMap<NormalizedRelayUrl, MutableMap<String, Long>>()
|
||||
|
||||
// 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<NormalizedRelayUrl, MutableSet<String>>()
|
||||
|
||||
// samples[relay][metric] -> ring buffer of latency in ms.
|
||||
private val samples =
|
||||
ConcurrentHashMap<NormalizedRelayUrl, ConcurrentHashMap<LatencyMetric, LatencyRingBuffer>>()
|
||||
|
||||
// ------ 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<NormalizedRelayUrl, RelayLatencySnapshot> {
|
||||
if (samples.isEmpty()) return persistentMapOf()
|
||||
val out = HashMap<NormalizedRelayUrl, RelayLatencySnapshot>(samples.size)
|
||||
for ((relay, perMetric) in samples) {
|
||||
val builder = HashMap<LatencyMetric, MetricSample>(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_<url>` Preferences key.
|
||||
*/
|
||||
fun samplesForPersistence(): Map<NormalizedRelayUrl, Map<LatencyMetric, IntArray>> {
|
||||
if (samples.isEmpty()) return emptyMap()
|
||||
val out = HashMap<NormalizedRelayUrl, Map<LatencyMetric, IntArray>>(samples.size)
|
||||
for ((relay, perMetric) in samples) {
|
||||
val inner = HashMap<LatencyMetric, IntArray>(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<NormalizedRelayUrl, Map<LatencyMetric, IntArray>>) {
|
||||
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<NormalizedRelayUrl, MutableMap<String, Long>>,
|
||||
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<String, Long> = java.util.Collections.synchronizedMap(java.util.LinkedHashMap<String, Long>(16, 0.75f, false))
|
||||
|
||||
private fun newConcurrentSet(): MutableSet<String> = java.util.Collections.newSetFromMap(ConcurrentHashMap<String, Boolean>())
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
+91
@@ -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())
|
||||
}
|
||||
}
|
||||
+239
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user