From d54e54b48ad3c5a49b2e74013fa6e29bc4df656a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 03:33:48 +0000 Subject: [PATCH] Add battle-tested sync accessories from vespa-relay's mirror Four pieces extracted from a production Nostr mirror (NosFabrica/vespa-relay), generalized to quartz's multiplatform primitives, with their test suites: - nip01Core.store.insertBisecting: batchInsert fails as a unit, so one bad event costs its whole batch (999 good events per bad one at a 1000-event batch). Bisecting isolates the offender in ~2*log2(n) extra writes, and a fixed write budget keeps a store-wide failure (full disk, dead engine) from turning one failed write into ~2n. - accessories.SyncBands: resume memory for fetchAllPages. Remembers the created_at band covered per (relay, filter) and asks only for the legs outside it, with inclusive edges so a page boundary cannot strand a run of same-second events. A finished negentropy reconcile records completeness through its start instant; a periodic full re-walk keeps stale claims from narrowing forever. Persistence is the caller's, via export/restore and an onChange hook. - accessories.PagingProgress: progress for paged walks measured on the time axis, the only axis whose end is known in advance - count-based percentages degenerate to downloaded/downloaded = 100%. Needs no COUNT support. - nip66RelayMonitor.reachability.HostStrikes: per-authority strike counting for outbox-scale fan-outs, where a filtering relay mints one url per user and per-url counters never converge. Ever-delivered overrides eviction in both race orders, and eviction surfaces exactly once for publishing. reachability.Unreachability (jvmAndroid): which failures may be published as a signed NIP-66 unreachable record - connection-level only, so a relay that answered the handshake and hung up mid-page is never libelled, and a caller's own bug is never the relay's fault. 57 tests pass on the JVM target; the common code uses quartz's ConcurrentMap, ConcurrentSet and TimeUtils only. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y4Pi9YYMhdzTFxRiV2jF9R --- .../client/accessories/PagingProgress.kt | 124 ++++++ .../relay/client/accessories/SyncBands.kt | 275 ++++++++++++++ .../quartz/nip01Core/store/BisectingInsert.kt | 89 +++++ .../reachability/HostStrikes.kt | 133 +++++++ .../client/accessories/PagingProgressTest.kt | 140 +++++++ .../relay/client/accessories/SyncBandsTest.kt | 355 ++++++++++++++++++ .../nip01Core/store/BisectingInsertTest.kt | 200 ++++++++++ .../reachability/HostStrikesTest.kt | 176 +++++++++ .../reachability/Unreachability.kt | 48 +++ .../reachability/UnreachabilityTest.kt | 70 ++++ 10 files changed, 1610 insertions(+) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgress.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBands.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsert.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikes.kt create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgressTest.kt create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBandsTest.kt create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsertTest.kt create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikesTest.kt create mode 100644 quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/Unreachability.kt create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/UnreachabilityTest.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgress.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgress.kt new file mode 100644 index 0000000000..028fa34979 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgress.kt @@ -0,0 +1,124 @@ +/* + * 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.quartz.nip01Core.relay.client.accessories + +import com.vitorpamplona.quartz.utils.TimeUtils +import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap +import kotlin.concurrent.Volatile + +/** + * How far a paged walk has got, measured on the time axis — the only axis + * whose end is known in advance. + * + * A paged fetch ([fetchAllPages]) has no event denominator: how many events + * exist is exactly what it is finding out, so every count-based percentage + * degenerates to `downloaded/downloaded = 100%`. The time axis has both ends + * before the first request — the filter's `until` (or now) down to its + * `since` (or [SyncBands.PLAUSIBLE_FLOOR]) — with each page's new `until` + * reporting the exact position between them. It needs no COUNT support. + * + * The estimate assumes events are spread evenly over time, which they are + * not — so it errs pessimistic on the tail, and is a bound, not a promise. + * + * One instance can serve many concurrent walks: keys are `"group|walk"`, and + * the group prefix scopes [fraction], [reached] and [etaMs] so two groups + * never report each other's numbers. + */ +class PagingProgress( + private val nowMillis: () -> Long = { TimeUtils.nowMillis() }, +) { + private class Walk( + val top: Long, + val bottom: Long, + val startedMs: Long, + @Volatile var current: Long, + ) + + private val walks = ConcurrentMap() + + /** Begin a walk over `[bottom, top]` seconds. An inverted window is not a walk. */ + fun begin( + key: String, + top: Long, + bottom: Long, + ) { + if (top > bottom) walks[key] = Walk(top, bottom, nowMillis(), top) + } + + /** The walk reached [until]; monotonic, so a page that jumps back cannot un-advance it. */ + fun mark( + key: String, + until: Long, + ) { + walks[key]?.let { + // Clamped to the walk's own floor: relays serve events stamped 0, + // and one of those would drag the position to the epoch. Below the + // floor means the walk is done, not time travel. + val reached = until.coerceAtLeast(it.bottom) + if (reached < it.current) it.current = reached + } + } + + fun finish(key: String) { + walks.remove(key) + } + + /** + * Fraction of the walk complete, averaged over every walk still going in + * [group] (or all of them when null) — averaged rather than summed + * because each covers its own span, so "half the walks done and half at + * zero" is 50%. + */ + fun fraction(group: String? = null): Double? { + val live = live(group) + if (live.isEmpty()) return null + return live.sumOf { w -> + val span = (w.top - w.bottom).coerceAtLeast(1) + ((w.top - w.current).toDouble() / span).coerceIn(0.0, 1.0) + } / live.size + } + + private fun live(group: String?): List = + if (group == null) { + walks.snapshot().values.toList() + } else { + walks + .snapshot() + .entries + .filter { it.key.startsWith("$group|") } + .map { it.value } + } + + /** The oldest second [group] has reached, or null when it is not walking. */ + fun reached(group: String? = null): Long? = live(group).minOfOrNull { it.current } + + /** Milliseconds left at the rate achieved so far, or null before it means anything. */ + fun etaMs(group: String? = null): Long? { + val f = fraction(group) ?: return null + // Under a few percent the extrapolation is dominated by connect time + // and produces numbers worse than saying nothing. + if (f < 0.02) return null + val oldestStart = live(group).minOfOrNull { it.startedMs } ?: return null + val elapsed = nowMillis() - oldestStart + if (elapsed < 5_000) return null + return ((elapsed / f) - elapsed).toLong() + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBands.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBands.kt new file mode 100644 index 0000000000..68bff2ca85 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBands.kt @@ -0,0 +1,275 @@ +/* + * 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.quartz.nip01Core.relay.client.accessories + +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.utils.TimeUtils +import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap + +/** + * How much of a filter's history has already been pulled from one relay, so a + * restart does not pull it again. + * + * A negentropy relay needs none of this — reconciliation downloads only the + * diff. Most relays lack NIP-77, and a paged fetch ([fetchAllPages]) has no + * memory: it re-downloads everything it walked last time, every restart, + * forever. So for those, remember the band of `created_at` covered per + * (relay, filter), and the next run asks only for the two legs outside it: + * + * stored band: |<-------- covered -------->| + * next fetch: <------| |------> + * + * Keyed by the WHOLE filter deliberately: any edit to a filter is a new key + * with no band, so the next run starts over — the safe direction to be wrong + * in, and the intended way to force a re-walk. + * + * A band does not guarantee completeness (a truncating relay, an event + * back-dated into a walked span). The trade is deliberate: re-reading a + * corpus on every restart is a certain daily cost, while both holes are + * occasional and self-heal on the next filter change or full re-walk. + * + * Persistence is the caller's: [export] the map on a schedule and [restore] + * it at startup. [onChange] fires whenever a band changes, so a persistence + * layer can mark itself dirty without polling. + */ +class SyncBands( + // How long a band may narrow work before the whole filter is walked + // again. Everything a band claims is a claim about the past; this is how + // long to trust it without re-testing. + private val fullResyncSeconds: Long = DEFAULT_FULL_RESYNC_SECONDS, + private val now: () -> Long = { TimeUtils.now() }, + private val onChange: () -> Unit = {}, +) { + /** + * What is already covered for one (relay, filter) pair. + * + * [complete] is the difference between "we walked this span" (a paged + * fetch) and "we are in sync below this point" (a finished negentropy + * reconcile, which compared the whole range). Only a complete band may + * skip its older leg. + * + * [fullAt] is when the last pass that started from nothing finished — the + * clock for the periodic re-walk. + */ + data class Band( + val minCreatedAt: Long, + val maxCreatedAt: Long, + val complete: Boolean = false, + val fullAt: Long = 0, + ) + + private val bands = ConcurrentMap() + + // filter -> its canonical json. Filter.toJson() runs to tens of thousands + // of characters for author-scoped filters, and a fan-out keys once per + // relay per cycle over the SAME handful of filter instances. Filter + // compares by identity, so this map is an identity cache; an + // equal-but-distinct filter still keys correctly, just without the cache. + private val fingerprints = ConcurrentMap() + + /** + * The filters to actually run now, given what is already covered: the + * whole filter when nothing is recorded (or the band went stale), + * otherwise the legs outside the band, clamped to the filter's own + * `since`/`until`. + * + * The legs are INCLUSIVE of the band's edges (`until = min`, not + * `min - 1`): a page boundary can split a run of events sharing one + * `created_at`, and excluding the edge would strand the rest of that + * second in no leg at all. The cost is re-reading one second's worth of + * events per leg, which a store rejects as duplicates. + */ + fun legs( + url: NormalizedRelayUrl, + filter: Filter, + ): List { + val band = bands[key(url, filter)] ?: return listOf(filter) + // Time for another full pass: relays gain old events, and without + // this the band's claim is never re-tested. + if (isStale(band)) return listOf(filter) + val legs = mutableListOf() + + // Older: up to and including the band's floor, but not past the + // filter's. A complete band has no older leg at all — the reconcile + // already compared the whole range. + if (!band.complete && (filter.since == null || band.minCreatedAt >= filter.since)) { + legs.add(filter.copy(until = minOf(band.minCreatedAt, filter.until ?: Long.MAX_VALUE))) + } + + // Newer: from the band's ceiling on, but not past the filter's. + if (filter.until == null || band.maxCreatedAt <= filter.until) { + legs.add(filter.copy(since = maxOf(band.maxCreatedAt, filter.since ?: Long.MIN_VALUE))) + } + return legs + } + + /** + * Widen the band for (url, filter) to include what a completed fetch saw. + * + * [paged] gates the mechanism: a negentropy sync needs no band, and + * recording one would only risk narrowing a future reconciliation. + * Nothing is recorded for a fetch that saw no events — an empty result + * says nothing about what the relay holds. + * + * [reconciledThrough] is the strong case: a FINISHED reconcile compared + * the filter's whole range, so the caller is in sync up to the instant + * the sync STARTED — recorded against that instant rather than the newest + * event seen, because "the relay had nothing newer" and "we never asked" + * must not look alike. + */ + fun record( + url: NormalizedRelayUrl, + filter: Filter, + observedMin: Long?, + observedMax: Long?, + paged: Boolean, + reconciledThrough: Long? = null, + ) { + if (reconciledThrough != null) { + put(url, filter, observedMin ?: reconciledThrough, reconciledThrough, complete = true) + return + } + if (!paged) return + // Guarded even though callers should filter with [isPlausible] per + // event: a 1970 floor or a far-future ceiling would make the band + // claim the whole timeline, and the leg outside it would ask for a + // range nothing can be in, forever. + if (observedMin == null || observedMax == null) return + if (!isPlausible(observedMin, now()) || !isPlausible(observedMax, now())) return + put(url, filter, observedMin, observedMax, complete = false) + } + + /** + * Widen (or reset) the band. A pass that ran because the previous band + * had gone stale REPLACES it: it re-walked the whole filter, so its own + * span is the complete picture and [Band.fullAt] restarts from here. + */ + private fun put( + url: NormalizedRelayUrl, + filter: Filter, + min: Long, + max: Long, + complete: Boolean, + ) { + val fresh = Band(min, max, complete, now()) + bands.merge(key(url, filter), fresh) { old, new -> + if (isStale(old)) { + new + } else { + Band( + minOf(old.minCreatedAt, new.minCreatedAt), + maxOf(old.maxCreatedAt, new.maxCreatedAt), + old.complete || new.complete, + old.fullAt, + ) + } + } + onChange() + } + + private fun isStale(band: Band): Boolean = now() - band.fullAt >= fullResyncSeconds + + /** + * The narrowest single filter that still covers what every one of [urls] + * needs — the window a shared negentropy snapshot has to be taken over. + * + * In steady state every relay carries a complete band and this collapses + * to `since = the oldest of their ceilings` — the difference between + * snapshotting an id set of millions and one of a few thousand. One relay + * that has never synced puts it back to the full filter, correctly: that + * relay genuinely needs everything. + */ + fun coveringWindow( + urls: List, + filter: Filter, + ): Filter { + if (urls.isEmpty()) return filter + var since = Long.MAX_VALUE + for (url in urls) { + val legs = legs(url, filter) + // More than one leg means an older gap this relay still wants, so + // the snapshot cannot start above the filter's own floor. + val only = legs.singleOrNull() ?: return filter + val legSince = only.since ?: return filter + since = minOf(since, legSince) + } + return if (since == Long.MAX_VALUE) filter else filter.copy(since = since) + } + + /** What is currently covered, for logging and tests. */ + fun band( + url: NormalizedRelayUrl, + filter: Filter, + ): Band? = bands[key(url, filter)] + + fun size(): Int = bands.size() + + /** A point-in-time copy of every band, for a persistence layer to write out. */ + fun export(): Map = bands.snapshot() + + /** Load previously [export]ed bands, e.g. at startup. */ + fun restore(entries: Map) { + for ((key, band) in entries) bands[key] = band + } + + /** + * The identity of one (relay, filter) pair. [Filter.toJson] is the + * protocol's own canonical form, so two filters that mean the same thing + * key the same way and any edit keys differently — exactly the "config + * changed, start over" rule. + */ + private fun key( + url: NormalizedRelayUrl, + filter: Filter, + ): String = "${url.url} ${fingerprints.getOrPut(filter) { filter.toJson() }}" + + companion object { + /** + * A week. Long enough that the narrow path is the normal one, short + * enough that anything a band is wrong about is wrong for days, not + * forever. + */ + const val DEFAULT_FULL_RESYNC_SECONDS = 7L * 24 * 60 * 60 + + /** + * 2020-01-01. Below this a `created_at` is a bug, not a date — the + * protocol did not exist. Also the natural floor for measuring a + * paged walk's progress when a filter names no `since`. + */ + const val PLAUSIBLE_FLOOR = 1_577_836_800L + + // Clock skew a relay may legitimately be ahead by. Past this, a + // created_at is the author's fiction rather than a time. + private const val FUTURE_SKEW_SECONDS = 86_400L + + /** + * Whether a `created_at` can be believed as evidence of coverage. + * Filter with this per EVENT, not over a leg's aggregate: one + * misdated event among hundreds of thousands would otherwise discard + * the whole relay's band. + */ + fun isPlausible( + createdAt: Long, + now: Long = TimeUtils.now(), + ): Boolean = createdAt in PLAUSIBLE_FLOOR..(now + FUTURE_SKEW_SECONDS) + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsert.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsert.kt new file mode 100644 index 0000000000..deae869a8d --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsert.kt @@ -0,0 +1,89 @@ +/* + * 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.quartz.nip01Core.store + +import com.vitorpamplona.quartz.nip01Core.core.Event +import kotlinx.coroutines.CancellationException + +/** + * Write [events] through [write]; if that throws, split the batch and write + * the halves, down to the single event the writer cannot take. + * + * A bulk write like [IEventStore.batchInsert] fails as a unit, so one bad + * event would otherwise cost the whole batch — 999 good events lost per bad + * one at a 1000-event batch, with no retry. Bisecting costs ~2·log2(n) extra + * writes on a failing batch, nothing on a healthy one, and ends holding the + * offender by itself for [onPoison] to report. Re-writing the good halves is + * safe: re-inserting an already-applied event is a duplicate the store + * rejects. + * + * Splitting assumes ONE event is at fault. When the store itself is refusing + * (a full disk, a dead engine) every half fails all the way down and + * isolation would turn one failed write into ~2n — precisely the wrong moment + * to multiply the load. So isolation spends a fixed [budget] of writes and + * hands the remainder to [onGaveUp]: "we could not say which" is a different + * fact from "this event is bad", and a caller should count them apart. + */ +suspend fun insertBisecting( + events: List, + write: suspend (List) -> List, + onOutcomes: (List) -> Unit, + onPoison: (Event, Throwable) -> Unit, + onGaveUp: (List, Throwable) -> Unit = { _, _ -> }, + budget: Int = ISOLATION_WRITE_BUDGET, +) = bisect(events, write, onOutcomes, onPoison, onGaveUp, intArrayOf(budget)) + +private suspend fun bisect( + events: List, + write: suspend (List) -> List, + onOutcomes: (List) -> Unit, + onPoison: (Event, Throwable) -> Unit, + onGaveUp: (List, Throwable) -> Unit, + budget: IntArray, +) { + if (events.isEmpty()) return + try { + onOutcomes(write(events)) + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + if (events.size == 1) { + onPoison(events.single(), e) + return + } + if (budget[0] <= 0) { + onGaveUp(events, e) + return + } + budget[0] -= 2 + val mid = events.size / 2 + bisect(events.subList(0, mid), write, onOutcomes, onPoison, onGaveUp, budget) + bisect(events.subList(mid, events.size), write, onOutcomes, onPoison, onGaveUp, budget) + } +} + +/** + * Writes one batch may spend isolating its bad events before giving up. + * Isolating k bad events out of n costs about `2·k·log2(n)` writes, so 64 + * covers three in a 1000-event batch — past the rate seen in practice. What + * it really bounds is the store-wide case, where every write fails. + */ +const val ISOLATION_WRITE_BUDGET = 64 diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikes.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikes.kt new file mode 100644 index 0000000000..11a2c2d51d --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikes.kt @@ -0,0 +1,133 @@ +/* + * 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.quartz.nip66RelayMonitor.reachability + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap +import com.vitorpamplona.quartz.utils.concurrent.ConcurrentSet + +/** + * Which relays are worth dialling, within one fan-out cycle. A discovered + * relay list is five figures of urls and most of them are corpses; without + * this, every cycle re-dials all of them and the working relays queue behind + * hosts that stopped existing years ago. + * + * Failures are counted per AUTHORITY (`host[:port]`), not per url: the outbox + * model mints one url per user for a filtering relay, so a per-url counter + * never reaches a threshold on any single one. The authority is host-only and + * does NOT fold a subdomain into its parent — those are different servers. + * + * [produced] overrides a strike race: a host that has ever delivered is never + * treated as dead for the rest of the cycle, whichever order the two events + * land in. Cycle-local; nothing persists — see [RelayReachabilityStore] for + * the part that survives a restart. + */ +class HostStrikes( + private val strikeLimit: Int = DEFAULT_STRIKE_LIMIT, + // Relays a previous run proved unreachable, and still within their TTL. + private val knownDead: Set = emptySet(), +) { + private val strikes = ConcurrentMap() + private val deadHosts = ConcurrentSet() + private val producedHosts = ConcurrentSet() + + private val delivered = ConcurrentSet() + private val failed = ConcurrentSet() + + /** Relays this cycle actually got something from — worth remembering as live. */ + val reachable: Set get() = delivered.snapshot() + + /** Relays this cycle could not reach at all. A relay that later delivered is not in it. */ + val unreachable: Set get() = failed.snapshot() - delivered.snapshot() + + /** + * Skip this relay? True when a previous run proved it dead (and no + * [produced] since), or when its whole authority has been struck out here. + */ + fun isDead(url: NormalizedRelayUrl): Boolean { + val authority = authorityOf(url.url) + if (authority in producedHosts) return false + return url in knownDead || authority in deadHosts + } + + /** + * This relay connected but delivered nothing before giving up. Count it + * against its authority and, at [strikeLimit], stop dialling the host. + * Returns the eviction — for the caller to publish — exactly when this + * strike is the one that took the host down: that is the only point where + * the evidence exists, because every sibling url is skipped without being + * dialled from here on. + */ + fun strike(url: NormalizedRelayUrl): Evicted? { + failed.add(url) + if (strikeLimit <= 0) return null + val authority = authorityOf(url.url) + if (authority in producedHosts || authority in deadHosts) return null + if (strikes.merge(authority, 1) { old, new -> old + new } < strikeLimit) return null + deadHosts.add(authority) + return Evicted(authority, strikeLimit) + } + + /** An authority struck out, and the evidence for it. */ + class Evicted( + val authority: String, + val strikes: Int, + ) + + /** This relay delivered. Its authority is alive, whatever else happened. */ + fun produced(url: NormalizedRelayUrl) { + delivered.add(url) + producedHosts.add(authorityOf(url.url)) + } + + /** For a cycle's closing line: how many hosts were dropped. */ + fun evictedHosts(): Int = deadHosts.size() + + fun summary(total: Int): String = + "${reachable.size} live, ${unreachable.size} unreachable, " + + "${deadHosts.size()} host(s) struck out, ${knownDead.size} skipped as known-dead of $total" + + companion object { + /** + * Three, because a single timeout is ordinary — a busy relay that + * never answered one REQ is not a dead one — while three separate + * urls on the same host all going silent is a server, not a + * coincidence. + */ + const val DEFAULT_STRIKE_LIMIT = 3 + + /** + * `host[:port]` — everything between the scheme and the first path + * slash. The port is part of it: two ports on one machine are two + * relays. + */ + fun authorityOf(url: String): String { + val afterScheme = + when { + url.startsWith("wss://") -> url.substring(6) + url.startsWith("ws://") -> url.substring(5) + else -> url + } + val slash = afterScheme.indexOf('/') + return if (slash >= 0) afterScheme.substring(0, slash) else afterScheme + } + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgressTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgressTest.kt new file mode 100644 index 0000000000..bf8a550c63 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgressTest.kt @@ -0,0 +1,140 @@ +/* + * 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.quartz.nip01Core.relay.client.accessories + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class PagingProgressTest { + private fun assertClose( + expected: Double, + actual: Double?, + message: String? = null, + ) { + assertTrue(actual != null && kotlin.math.abs(expected - actual) < 0.001, "${message ?: ""} expected $expected got $actual") + } + + @Test + fun `progress is the walked share of the time window`() { + val p = PagingProgress() + p.begin("a", top = 1_000L, bottom = 0L) + + assertClose(0.0, p.fraction(), "nothing walked yet") + + p.mark("a", 750L) + assertClose(0.25, p.fraction()) + + p.mark("a", 100L) + assertClose(0.90, p.fraction()) + } + + @Test + fun `a page that jumps backwards cannot un-advance the walk`() { + // Pages arrive from one relay in order, but nothing in the protocol + // guarantees it, and a percentage that goes DOWN is worse than one that + // is slightly wrong — it reads as the sync having lost ground. + val p = PagingProgress() + p.begin("a", top = 1_000L, bottom = 0L) + + p.mark("a", 200L) + p.mark("a", 900L) + + assertClose(0.80, p.fraction(), "the later higher until is ignored") + } + + @Test + fun `walks average rather than sum`() { + // Two relays each walking their own window: one done and one untouched + // is half way — not 100% as summing would give. + val p = PagingProgress() + p.begin("a", top = 1_000L, bottom = 0L) + p.begin("b", top = 500L, bottom = 0L) + + p.mark("a", 0L) + + assertClose(0.5, p.fraction()) + } + + @Test + fun `a finished walk leaves the average`() { + val p = PagingProgress() + p.begin("a", top = 1_000L, bottom = 0L) + p.begin("b", top = 1_000L, bottom = 0L) + p.mark("b", 500L) + + p.finish("a") + + assertClose(0.5, p.fraction(), "only b is still walking") + p.finish("b") + assertNull(p.fraction(), "nothing walking means no number to report") + } + + @Test + fun `a group prefix scopes the numbers to its own walks`() { + // One instance serves many concurrent walks; without the scope two + // streams would print each other's percentages. + val p = PagingProgress() + p.begin("streamA|wss://r1", top = 1_000L, bottom = 0L) + p.begin("streamB|wss://r2", top = 1_000L, bottom = 0L) + p.mark("streamA|wss://r1", 0L) + + assertClose(1.0, p.fraction("streamA")) + assertClose(0.0, p.fraction("streamB")) + assertClose(0.5, p.fraction()) + } + + @Test + fun `an inverted or empty window is not a walk`() { + // A leg whose since is above its until asks for a range nothing can be + // in. Dividing by that span would produce infinities on the status line. + val p = PagingProgress() + + p.begin("a", top = 100L, bottom = 900L) + + assertNull(p.fraction()) + } + + @Test + fun `no ETA before the estimate means anything`() { + val p = PagingProgress() + p.begin("a", top = 1_000_000L, bottom = 0L) + + p.mark("a", 999_000L) + + // 0.1% in: extrapolating here yields days-long ETAs from connect + // latency alone, which is worse than printing nothing. + assertNull(p.etaMs(), "too early to extrapolate") + } + + @Test + fun `ETA extrapolates from the rate achieved so far`() { + var clock = 1_000_000L + val p = PagingProgress(nowMillis = { clock }) + p.begin("a", top = 1_000L, bottom = 0L) + p.mark("a", 500L) + + // Half way after six seconds: whatever has elapsed is also what remains. + clock += 6_000 + assertEquals(6_000L, p.etaMs()) + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBandsTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBandsTest.kt new file mode 100644 index 0000000000..d0ffe5290a --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBandsTest.kt @@ -0,0 +1,355 @@ +/* + * 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.quartz.nip01Core.relay.client.accessories + +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertSame +import kotlin.test.assertTrue + +/** + * A paged relay has no memory of what it already sent, so without a band every + * restart re-downloads its whole corpus. These pin the band arithmetic and, more + * importantly, the cases where a band must NOT be used — a stale band silently + * skips events, which is a worse failure than re-reading them. + */ +class SyncBandsTest { + private val relay = RelayUrlNormalizer.normalize("wss://relay.example") + private val other = RelayUrlNormalizer.normalize("wss://other.example") + private val profiles = Filter(kinds = listOf(0)) + + private fun now(): Long = TimeUtils.now() + + // ---- the band arithmetic ---------------------------------------------- + + @Test + fun `with nothing recorded the whole filter is fetched`() { + val c = SyncBands() + assertEquals(listOf(profiles), c.legs(relay, profiles)) + } + + @Test + fun `a recorded band is fetched around rather than through`() { + val c = SyncBands() + c.record(relay, profiles, observedMin = 1_700_001_000L, observedMax = 1_700_002_000L, paged = true) + + val legs = c.legs(relay, profiles) + assertEquals(2, legs.size, "one leg older than the band and one newer") + assertEquals(1_700_001_000L, legs[0].until, "older leg stops AT the band floor") + assertNull(legs[0].since, "and reaches as far back as the filter allows") + assertEquals(1_700_002_000L, legs[1].since, "newer leg starts AT its ceiling") + assertNull(legs[1].until) + } + + @Test + fun `an event sharing the band boundary second is still reachable`() { + // A paged relay cuts pages by count, so a boundary can fall inside a run + // of events sharing one created_at. Excluding the edge would strand the + // rest of that second in no leg at all, while the band called it covered. + val c = SyncBands() + c.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = true) + + val legs = c.legs(relay, profiles) + + fun reachable(t: Long) = legs.any { (it.since ?: Long.MIN_VALUE) <= t && t <= (it.until ?: Long.MAX_VALUE) } + + assertTrue(reachable(1_700_001_000L), "the band floor second must be re-read") + assertTrue(reachable(1_700_002_000L), "and its ceiling second") + assertTrue(reachable(1_700_000_999L), "below the band") + assertTrue(reachable(1_700_002_001L), "above it") + // Only the interior is skipped, which is the entire point. + assertTrue(!reachable(1_700_001_500L), "the covered interior is not re-read") + } + + @Test + fun `successive runs widen the band rather than replacing it`() { + val c = SyncBands() + c.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = true) + // A later run reaches further back and picks up newer events. + c.record(relay, profiles, 1_700_000_500L, 1_700_002_500L, paged = true) + + val band = c.band(relay, profiles)!! + assertEquals(1_700_000_500L, band.minCreatedAt) + assertEquals(1_700_002_500L, band.maxCreatedAt) + } + + @Test + fun `a capped relay walks further back on each run`() { + // The case that makes this worth having: a relay that only ever answers + // with its newest N events. Each run starts below the last one's floor. + val c = SyncBands() + c.record(relay, profiles, 1_700_009_000L, 1_700_010_000L, paged = true) + assertEquals(1_700_009_000L, c.legs(relay, profiles)[0].until) + + c.record(relay, profiles, 1_700_008_000L, 1_700_008_999L, paged = true) + assertEquals(1_700_008_000L, c.legs(relay, profiles)[0].until) + } + + // ---- when a band must not be used -------------------------------------- + + @Test + fun `a negentropy sync that reported no outcome records nothing`() { + // Only a sync that says how far it reconciled earns a band; a bare + // paged=false call carries no claim to record. + val c = SyncBands() + c.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = false) + assertNull(c.band(relay, profiles)) + assertEquals(listOf(profiles), c.legs(relay, profiles)) + } + + // ---- coverage: what a finished reconcile earns ------------------------- + + @Test + fun `a finished reconcile is in sync through the instant it started`() { + // Not through the newest event it happened to see: "the relay had nothing + // newer" and "we never asked" must not record the same thing. + val c = SyncBands() + val startedAt = now() - 60 + c.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = false, reconciledThrough = startedAt) + + val band = c.band(relay, profiles)!! + assertTrue(band.complete) + assertEquals(startedAt, band.maxCreatedAt) + } + + @Test + fun `a reconcile that downloaded nothing still records coverage`() { + // The empty case is the WHOLE point: nothing came back because we already + // have it, and that is exactly when the next run should ask for a sliver. + val c = SyncBands() + val startedAt = now() - 60 + c.record(relay, profiles, null, null, paged = false, reconciledThrough = startedAt) + + val leg = c.legs(relay, profiles).single() + assertEquals(startedAt, leg.since) + assertNull(leg.until) + } + + @Test + fun `a complete band drops its older leg while a paged one keeps it`() { + val reconciled = SyncBands() + reconciled.record(relay, profiles, null, null, paged = false, reconciledThrough = 1_700_002_000L) + val only = reconciled.legs(relay, profiles).single() + assertEquals(1_700_002_000L, only.since) + + val walked = SyncBands() + walked.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = true) + assertEquals(2, walked.legs(relay, profiles).size, "a paged walk says nothing about what it never asked for") + } + + // ---- the periodic full re-walk ----------------------------------------- + + @Test + fun `a band stops narrowing once it is older than the resync period`() { + val c = SyncBands(fullResyncSeconds = 60) + c.record(relay, profiles, null, null, paged = false, reconciledThrough = now() - 3600) + // Recorded 'now' whatever the created_at claim, so age it by rewriting. + c.record(relay, profiles, null, null, paged = false, reconciledThrough = now()) + assertEquals(1, c.legs(relay, profiles).size, "fresh band still narrows") + + val stale = SyncBands(fullResyncSeconds = 0) + stale.record(relay, profiles, null, null, paged = false, reconciledThrough = now()) + assertSame(profiles, stale.legs(relay, profiles).single(), "a band past its period re-walks everything") + } + + @Test + fun `the re-walk replaces the old claim instead of widening it`() { + // Widening would carry the stale band's floor forward forever and the + // periodic pass would never actually reset anything. + val c = SyncBands(fullResyncSeconds = 0) + c.record(relay, profiles, 1_700_000_000L, 1_700_001_000L, paged = true) + c.record(relay, profiles, 1_700_005_000L, 1_700_006_000L, paged = true) + + val band = c.band(relay, profiles)!! + assertEquals(1_700_005_000L, band.minCreatedAt, "the second pass walked everything; its span is the whole picture") + } + + // ---- the shared snapshot window ---------------------------------------- + + @Test + fun `covering window collapses to the oldest ceiling once everyone is caught up`() { + val c = SyncBands() + c.record(relay, profiles, null, null, paged = false, reconciledThrough = 1_700_009_000L) + c.record(other, profiles, null, null, paged = false, reconciledThrough = 1_700_003_000L) + + assertEquals(1_700_003_000L, c.coveringWindow(listOf(relay, other), profiles).since) + } + + @Test + fun `one relay that has never synced puts the window back to the whole filter`() { + // It genuinely needs everything — narrowing the shared snapshot would + // reconcile it against ids we never looked up. + val c = SyncBands() + c.record(relay, profiles, null, null, paged = false, reconciledThrough = 1_700_009_000L) + + // The filter itself, unnarrowed — identity, since Filter has no equals. + assertSame(profiles, c.coveringWindow(listOf(relay, other), profiles)) + assertSame(profiles, c.coveringWindow(emptyList(), profiles)) + } + + @Test + fun `one shared window serves a whole stream of relays`() { + // Every url in a stream shares that stream's filter, so a backfill can + // take ONE snapshot for all of them instead of walking the identical + // range once per relay for byte-identical answers. + val c = SyncBands() + val third = RelayUrlNormalizer.normalize("wss://third.example") + c.record(relay, profiles, null, null, paged = false, reconciledThrough = 1_700_009_000L) + c.record(other, profiles, null, null, paged = false, reconciledThrough = 1_700_003_000L) + c.record(third, profiles, null, null, paged = false, reconciledThrough = 1_700_007_000L) + + // The hungriest of them sets the floor; the other two re-read a little. + assertEquals(1_700_003_000L, c.coveringWindow(listOf(relay, other, third), profiles).since) + } + + @Test + fun `a relay with an older gap also widens the shared window`() { + val c = SyncBands() + c.record(relay, profiles, null, null, paged = false, reconciledThrough = 1_700_009_000L) + c.record(other, profiles, 1_700_001_000L, 1_700_002_000L, paged = true) + + assertSame(profiles, c.coveringWindow(listOf(relay, other), profiles)) + } + + @Test + fun `an empty fetch records nothing`() { + // No events says nothing about what the relay holds, only that this + // window was empty — recording it would fabricate coverage. + val c = SyncBands() + c.record(relay, profiles, null, null, paged = true) + assertNull(c.band(relay, profiles)) + } + + @Test + fun `one misdated event does not cost a relay its whole band`() { + // A single future-dated stamp among hundreds of thousands must not fail + // a check applied to the aggregate. Screening per event keeps the rest. + val c = SyncBands() + val far = now() + 400L * 86_400 + val observed = listOf(1_700_001_000L, far, 1_700_002_000L, 0L) + + val plausible = observed.filter { SyncBands.isPlausible(it) } + c.record(relay, profiles, plausible.min(), plausible.max(), paged = true) + + val band = c.band(relay, profiles)!! + assertEquals(1_700_001_000L, band.minCreatedAt) + assertEquals(1_700_002_000L, band.maxCreatedAt) + } + + @Test + fun `changing the filter starts over`() { + val c = SyncBands() + c.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = true) + + // Widening the kinds means the old band skipped events it never fetched. + val wider = Filter(kinds = listOf(0, 10002)) + assertEquals(listOf(wider), c.legs(relay, wider), "a new filter has no band") + assertNull(c.band(relay, wider)) + // ...and the original is untouched, so reverting resumes where it was. + assertEquals(1_700_001_000L, c.band(relay, profiles)!!.minCreatedAt) + } + + @Test + fun `each relay keeps its own band`() { + val c = SyncBands() + c.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = true) + assertEquals(listOf(profiles), c.legs(other, profiles)) + } + + // ---- the filter's own bounds still win --------------------------------- + + @Test + fun `a bounded filter never widens past its own since and until`() { + val bounded = Filter(kinds = listOf(0), since = 1_700_001_000L, until = 1_700_005_000L) + val c = SyncBands() + c.record(relay, bounded, 1_700_002_000L, 1_700_003_000L, paged = true) + + val legs = c.legs(relay, bounded) + assertEquals(2, legs.size) + assertEquals(1_700_001_000L, legs[0].since, "the older leg keeps the configured floor") + assertEquals(1_700_002_000L, legs[0].until) + assertEquals(1_700_003_000L, legs[1].since) + assertEquals(1_700_005_000L, legs[1].until, "the newer leg keeps the configured ceiling") + } + + @Test + fun `a fully covered bounded filter re-reads only its two edge seconds`() { + // Inclusive edges mean "covered" can never quite mean "ask for nothing": + // the two boundary seconds are always re-read, because that is the only + // way to catch a run of same-second events a page boundary cut in half. + val bounded = Filter(kinds = listOf(0), since = 1_700_001_000L, until = 1_700_005_000L) + val c = SyncBands() + c.record(relay, bounded, 1_700_001_000L, 1_700_005_000L, paged = true) + + val legs = c.legs(relay, bounded) + assertEquals(2, legs.size) + assertEquals(1_700_001_000L to 1_700_001_000L, legs[0].since to legs[0].until, "the floor second only") + assertEquals(1_700_005_000L to 1_700_005_000L, legs[1].since to legs[1].until, "the ceiling second only") + } + + // ---- persistence hooks -------------------------------------------------- + + @Test + fun `export and restore round-trip the bands`() { + val c = SyncBands() + c.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = true) + + val reopened = SyncBands() + reopened.restore(c.export()) + + val band = reopened.band(relay, profiles)!! + assertEquals(1_700_001_000L, band.minCreatedAt) + assertEquals(1_700_002_000L, band.maxCreatedAt) + } + + @Test + fun `onChange fires when a band changes so persistence can mark dirty`() { + var changes = 0 + val c = SyncBands(onChange = { changes++ }) + + c.record(relay, profiles, null, null, paged = true) + assertEquals(0, changes, "an empty fetch records nothing and must not dirty the store") + + c.record(relay, profiles, 1_700_001_000L, 1_700_002_000L, paged = true) + assertEquals(1, changes) + } + + @Test + fun `the same filter instance is fingerprinted once`() { + // Filter.toJson() runs to tens of thousands of characters for an + // author-scoped filter, and a fan-out keys once per relay per cycle. + val big = Filter(kinds = listOf(30382), authors = (1..500).map { it.toString(16).padStart(64, '0') }) + val c = SyncBands() + c.record(relay, big, 1_700_001_000L, 1_700_002_000L, paged = true) + + // Same instance, many lookups: still one band, and cheap. + repeat(50) { c.legs(relay, big) } + assertEquals(1_700_001_000L, c.band(relay, big)!!.minCreatedAt) + + // An equal-but-distinct instance keys the same way; it just misses the cache. + val copy = Filter(kinds = listOf(30382), authors = (1..500).map { it.toString(16).padStart(64, '0') }) + assertEquals(1_700_001_000L, c.band(relay, copy)?.minCreatedAt, "identity caching must not change the key") + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsertTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsertTest.kt new file mode 100644 index 0000000000..2b813995c9 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsertTest.kt @@ -0,0 +1,200 @@ +/* + * 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.quartz.nip01Core.store + +import com.vitorpamplona.quartz.nip01Core.core.Event +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertTrue + +/** + * A bulk write fails as a unit, so without isolation one event the store cannot + * take costs its whole batch — 999 good events per bad one at the default size, + * dropped silently and counted as a multiple of the batch rather than as a number + * of bad events. These pin the isolation that stops that. + */ +class BisectingInsertTest { + private fun event(n: Int) = + Event( + id = n.toString().padStart(64, '0'), + pubKey = "a1".repeat(32), + createdAt = 1_700_000_000L + n, + kind = 1, + tags = emptyArray(), + content = "e$n", + sig = "b2".repeat(32), + ) + + /** Accepts everything except the named ids, which make the whole write throw. */ + private class Writer( + private val poison: Set, + ) { + val calls = mutableListOf() + var eventsWritten = 0 + + suspend fun write(batch: List): List { + calls.add(batch.size) + batch.firstOrNull { it.id in poison }?.let { + throw IndexOutOfBoundsException("Index: 1 Size: 1") + } + eventsWritten += batch.size + return batch.map { IEventStore.InsertOutcome.Accepted } + } + } + + private val gaveUp = mutableListOf() + + private suspend fun run( + events: List, + poison: Set, + ): Triple>> { + val writer = Writer(poison) + var accepted = 0 + val poisoned = mutableListOf>() + gaveUp.clear() + insertBisecting( + events = events, + write = { writer.write(it) }, + onOutcomes = { accepted += it.size }, + onPoison = { e, t -> poisoned.add(e to t) }, + onGaveUp = { batch, _ -> gaveUp.add(batch.size) }, + ) + return Triple(writer, accepted, poisoned) + } + + @Test + fun `a healthy batch is written once and costs nothing extra`() = + runTest { + val events = (1..64).map(::event) + val (writer, accepted, poisoned) = run(events, emptySet()) + + assertEquals(listOf(64), writer.calls, "no bisection on a batch that works") + assertEquals(64, accepted) + assertTrue(poisoned.isEmpty()) + } + + @Test + fun `one poison event costs only itself and not the batch`() = + runTest { + val events = (1..64).map(::event) + val bad = events[37].id + val (_, accepted, poisoned) = run(events, setOf(bad)) + + // This is the whole point: 63 of 64 still land. + assertEquals(63, accepted, "every event except the poison one must still be written") + assertEquals(1, poisoned.size) + assertEquals(bad, poisoned.single().first.id, "the isolated event is the one that throws") + assertTrue(poisoned.single().second is IndexOutOfBoundsException) + } + + @Test + fun `isolating stays logarithmic instead of falling back to one-by-one`() = + runTest { + val events = (1..1024).map(::event) + val (writer, accepted, _) = run(events, setOf(events[500].id)) + + assertEquals(1023, accepted) + // ~2*log2(n) writes, nowhere near the 1024 a per-event fallback would cost. + assertTrue(writer.calls.size < 32, "expected a logarithmic split, got ${writer.calls.size} writes") + } + + @Test + fun `several poison events are each isolated`() = + runTest { + val events = (1..64).map(::event) + val bad = setOf(events[0].id, events[31].id, events[63].id) + val (_, accepted, poisoned) = run(events, bad) + + assertEquals(61, accepted) + assertEquals(bad, poisoned.map { it.first.id }.toSet()) + } + + @Test + fun `a store-wide failure gives up instead of splitting all the way down`() = + runTest { + // Everything fails — a full disk, a dead engine. Splitting to singletons + // would cost ~2n writes at the worst possible moment. + val events = (1..1024).map(::event) + val (writer, accepted, poisoned) = run(events, events.map { it.id }.toSet()) + + assertEquals(0, accepted) + assertTrue( + writer.calls.size < 100, + "a store-wide failure must not cost ~2n writes; spent ${writer.calls.size}", + ) + // Nothing is silently lost: whatever isolation could not name is still + // handed back, so the caller can count it. + assertEquals(1024, poisoned.size + gaveUp.sum(), "every event must be accounted for") + assertTrue(gaveUp.isNotEmpty(), "the remainder should be reported as unisolated") + } + + @Test + fun `the budget is spent isolating rather than hoarded`() = + runTest { + // One bad event in 1024 must still be found — the guard bounds the + // pathological case without breaking the case it was built for. + val events = (1..1024).map(::event) + val (_, accepted, poisoned) = run(events, setOf(events[900].id)) + + assertEquals(1023, accepted) + assertEquals(events[900].id, poisoned.single().first.id) + assertTrue(gaveUp.isEmpty(), "a single bad event fits well inside the budget") + } + + @Test + fun `a batch of nothing but poison loses nothing else`() = + runTest { + val events = (1..4).map(::event) + val (_, accepted, poisoned) = run(events, events.map { it.id }.toSet()) + + assertEquals(0, accepted) + assertEquals(4, poisoned.size, "each one named, rather than one count of four") + } + + @Test + fun `cancellation propagates instead of being mistaken for a poison event`() = + runTest { + // Shutdown cancels the ingest scope mid-write. Treating that as "this + // event is bad" would drop good events and keep bisecting while the + // caller is trying to stop. + val events = (1..16).map(::event) + assertFailsWith { + insertBisecting( + events = events, + write = { throw CancellationException("shutting down") }, + onOutcomes = { }, + onPoison = { _, _ -> error("cancellation must not be reported as poison") }, + ) + } + } + + @Test + fun `an empty batch is a no-op`() = + runTest { + val (writer, accepted, poisoned) = run(emptyList(), emptySet()) + assertTrue(writer.calls.isEmpty()) + assertEquals(0, accepted) + assertTrue(poisoned.isEmpty()) + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikesTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikesTest.kt new file mode 100644 index 0000000000..d86141992a --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikesTest.kt @@ -0,0 +1,176 @@ +/* + * 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.quartz.nip66RelayMonitor.reachability + +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * An outbox list is five figures of urls and mostly corpses. These pin the two + * rules that make the difference: failures count per HOST (or a filtering + * relay's hundreds of per-user urls never add up to anything), and a host that + * has ever delivered is never dropped (or a fan-out this wide sheds working + * relays on a race). + */ +class HostStrikesTest { + private fun url(u: String) = RelayUrlNormalizer.normalize(u) + + // ---- the authority key -------------------------------------------------- + + @Test + fun `per-user path urls on one host share an authority`() { + val a = HostStrikes.authorityOf("wss://filter.nostr.wine/npub1aaaa?broadcast=true") + val b = HostStrikes.authorityOf("wss://filter.nostr.wine/npub1bbbb") + assertEquals("filter.nostr.wine", a) + assertEquals(a, b) + assertEquals(a, HostStrikes.authorityOf("wss://filter.nostr.wine")) + } + + @Test + fun `a subdomain is not folded into its parent`() { + // Different servers. Shedding the filtering one must never take out the + // bare host, which may be perfectly open. + assertTrue( + HostStrikes.authorityOf("wss://filter.nostr.wine/npub1x") != + HostStrikes.authorityOf("wss://nostr.wine"), + ) + } + + @Test + fun `the port is part of the authority`() { + assertEquals("relay.example.com:443", HostStrikes.authorityOf("wss://relay.example.com:443/npub1z")) + assertTrue( + HostStrikes.authorityOf("wss://example.com:443") != HostStrikes.authorityOf("wss://example.com:8080"), + ) + } + + // ---- striking ----------------------------------------------------------- + + @Test + fun `one host is struck out by failures spread across its many urls`() { + // The whole point: no single url would ever reach the threshold alone. + val h = HostStrikes() + h.strike(url("wss://filter.example/npub1aaa")) + h.strike(url("wss://filter.example/npub1bbb")) + assertFalse(h.isDead(url("wss://filter.example/npub1ccc")), "two strikes is not yet a verdict") + + h.strike(url("wss://filter.example/npub1ccc")) + assertTrue(h.isDead(url("wss://filter.example/npub1ddd")), "the host is out — including urls never tried") + } + + @Test + fun `eviction returns a verdict exactly once for publishing`() { + // The eviction is the only finding that will ever exist about the sibling + // urls under this host: from here they are skipped without being dialled, + // so nothing observes them again. It must surface exactly once — silent + // would publish nothing, repeated would rewrite the record every strike. + val h = HostStrikes() + assertNull(h.strike(url("wss://filter.example/npub1")), "one strike is not a verdict") + assertNull(h.strike(url("wss://filter.example/npub2")), "two is not either") + + val evicted = h.strike(url("wss://filter.example/npub3")) + assertNotNull(evicted, "the third strike is the finding") + assertEquals("filter.example", evicted.authority) + assertEquals(3, evicted.strikes) + + assertNull(h.strike(url("wss://filter.example/npub4")), "already evicted — do not report it again") + } + + @Test + fun `a host that has delivered is never evicted so nothing is published`() { + val h = HostStrikes() + h.produced(url("wss://busy.example/npubY")) + repeat(5) { assertNull(h.strike(url("wss://busy.example/npub$it")), "ever-produced outranks any strike") } + } + + @Test + fun `striking one host leaves every other alone`() { + val h = HostStrikes() + repeat(5) { h.strike(url("wss://filter.example/npub$it")) } + assertTrue(h.isDead(url("wss://filter.example/npub1"))) + assertFalse(h.isDead(url("wss://example.com"))) + assertFalse(h.isDead(url("wss://other.example"))) + } + + @Test + fun `a host that ever delivered is never dead whichever way the race lands`() { + // At a hundred relays in flight one worker can strike an authority out at + // the same instant another is receiving from it. Ever-produced must win in + // both orders, which is why it overrides rather than clearing strikes. + val strikeFirst = HostStrikes() + repeat(3) { strikeFirst.strike(url("wss://busy.example/npub$it")) } + assertTrue(strikeFirst.isDead(url("wss://busy.example/npubX"))) + strikeFirst.produced(url("wss://busy.example/npubY")) + assertFalse(strikeFirst.isDead(url("wss://busy.example/npubX")), "a delivery revives the whole host") + + val produceFirst = HostStrikes() + produceFirst.produced(url("wss://busy.example/npubY")) + repeat(5) { produceFirst.strike(url("wss://busy.example/npub$it")) } + assertFalse(produceFirst.isDead(url("wss://busy.example/npubX")), "later strikes cannot bury it") + } + + @Test + fun `a zero strike limit disables eviction entirely`() { + val h = HostStrikes(strikeLimit = 0) + repeat(50) { h.strike(url("wss://filter.example/npub$it")) } + assertFalse(h.isDead(url("wss://filter.example/npub1"))) + } + + // ---- what a previous run already learned --------------------------------- + + @Test + fun `a relay a previous run proved dead is skipped without dialling`() { + val gone = url("wss://gone.example") + val h = HostStrikes(knownDead = setOf(gone)) + assertTrue(h.isDead(gone)) + assertFalse(h.isDead(url("wss://alive.example"))) + } + + @Test + fun `a known-dead relay that answers anyway is believed over the record`() { + // Relays come back. A TTL'd record is "not now" and never "never again": + // a delivery this cycle must beat what an earlier one wrote down. + val back = url("wss://back.example") + val h = HostStrikes(knownDead = setOf(back)) + h.produced(back) + assertFalse(h.isDead(back)) + } + + // ---- what gets written back --------------------------------------------- + + @Test + fun `only relays actually dialled are reported and delivery clears a failure`() { + val h = HostStrikes() + val good = url("wss://good.example") + val bad = url("wss://bad.example") + h.strike(bad) + h.strike(good) + h.produced(good) + + assertEquals(setOf(good), h.reachable) + assertEquals(setOf(bad), h.unreachable, "a relay that later delivered is not reported dead") + } +} diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/Unreachability.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/Unreachability.kt new file mode 100644 index 0000000000..a9b1cf634e --- /dev/null +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/Unreachability.kt @@ -0,0 +1,48 @@ +/* + * 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.quartz.nip66RelayMonitor.reachability + +/** + * Whether a failure may be published as "this relay is unreachable". + * + * The distinction matters because the answer is PUBLISHED: a negative NIP-66 + * record is a signed, public statement about someone else's server. A relay + * that completes a handshake and then hangs up mid-page is emphatically + * reachable, and an exception thrown by the caller's own code says nothing + * about the relay at all. So this asks only about the connection itself — + * name resolution, routing, refusal, TLS. + * + * Unknown failures stay quiet: the cost of silence is one retry next cycle, + * the cost of being wrong is a false record carrying the monitor's signature. + */ +object Unreachability { + fun proves(e: Exception): Boolean = + when (e) { + is java.net.UnknownHostException, + is java.net.ConnectException, + is java.net.NoRouteToHostException, + is java.net.PortUnreachableException, + is javax.net.ssl.SSLHandshakeException, + -> true + + else -> false + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/UnreachabilityTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/UnreachabilityTest.kt new file mode 100644 index 0000000000..066f4b4b67 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/UnreachabilityTest.kt @@ -0,0 +1,70 @@ +/* + * 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.quartz.nip66RelayMonitor.reachability + +import java.io.EOFException +import java.net.ConnectException +import java.net.SocketTimeoutException +import java.net.UnknownHostException +import javax.net.ssl.SSLHandshakeException +import kotlin.test.Test +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +/** + * What a monitor is willing to SAY about someone else's relay. A negative + * NIP-66 record is signed and public, so only failures of the connection + * itself may be published as "unreachable". + */ +class UnreachabilityTest { + private fun proves(e: Exception) = Unreachability.proves(e) + + @Test + fun `a connection that never opened is unreachable`() { + assertTrue(proves(UnknownHostException("no such host"))) + assertTrue(proves(ConnectException("connection refused"))) + assertTrue(proves(SSLHandshakeException("cert expired"))) + } + + @Test + fun `a relay that hung up mid-transfer is not unreachable`() { + // A relay that answers the handshake in 50ms and then sends EOFException + // part-way through a large page is reachable; it declined to finish a + // query. Publishing "unreachable" would be a false statement about a + // working server. + assertFalse(proves(EOFException("stream closed"))) + } + + @Test + fun `our own bug is never the relay's fault`() { + assertFalse(proves(ConcurrentModificationException())) + assertFalse(proves(NullPointerException())) + assertFalse(proves(ClassCastException("HashMap\$Node cannot be cast"))) + } + + @Test + fun `an unrecognised failure stays quiet`() { + // Conservative on purpose: staying quiet costs one retry next cycle, + // being wrong costs a false record carrying the monitor's signature. + assertFalse(proves(SocketTimeoutException("read timed out"))) + assertFalse(proves(RuntimeException("something new"))) + } +}