laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-08 23:54:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add battle-tested sync accessories from vespa-relay's mirror

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y4Pi9YYMhdzTFxRiV2jF9R
This commit is contained in:
Claude
2026-08-04 03:33:48 +00:00
parent ba78eb599e
commit d54e54b48a
10 changed files with 1610 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgress.kt
+124
View File
@@ -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<String, Walk>()
/** 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<Walk> =
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()
}
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBands.kt
+275
View File
@@ -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<String, Band>()
// 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<Filter, String>()
/**
* 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<Filter> {
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<Filter>()
// 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<NormalizedRelayUrl>,
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<String, Band> = bands.snapshot()
/** Load previously [export]ed bands, e.g. at startup. */
fun restore(entries: Map<String, Band>) {
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)
}
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsert.kt
+89
View File
@@ -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<Event>,
write: suspend (List<Event>) -> List<IEventStore.InsertOutcome>,
onOutcomes: (List<IEventStore.InsertOutcome>) -> Unit,
onPoison: (Event, Throwable) -> Unit,
onGaveUp: (List<Event>, Throwable) -> Unit = { _, _ -> },
budget: Int = ISOLATION_WRITE_BUDGET,
) = bisect(events, write, onOutcomes, onPoison, onGaveUp, intArrayOf(budget))
private suspend fun bisect(
events: List<Event>,
write: suspend (List<Event>) -> List<IEventStore.InsertOutcome>,
onOutcomes: (List<IEventStore.InsertOutcome>) -> Unit,
onPoison: (Event, Throwable) -> Unit,
onGaveUp: (List<Event>, 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
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikes.kt
+133
View File
@@ -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<NormalizedRelayUrl> = emptySet(),
) {
private val strikes = ConcurrentMap<String, Int>()
private val deadHosts = ConcurrentSet<String>()
private val producedHosts = ConcurrentSet<String>()
private val delivered = ConcurrentSet<NormalizedRelayUrl>()
private val failed = ConcurrentSet<NormalizedRelayUrl>()
/** Relays this cycle actually got something from — worth remembering as live. */
val reachable: Set<NormalizedRelayUrl> get() = delivered.snapshot()
/** Relays this cycle could not reach at all. A relay that later delivered is not in it. */
val unreachable: Set<NormalizedRelayUrl> 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
}
}
}
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/PagingProgressTest.kt
+140
View File
@@ -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())
}
}
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncBandsTest.kt
+355
View File
@@ -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")
}
}
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/BisectingInsertTest.kt
+200
View File
@@ -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<String>,
) {
val calls = mutableListOf<Int>()
var eventsWritten = 0
suspend fun write(batch: List<Event>): List<IEventStore.InsertOutcome> {
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<Int>()
private suspend fun run(
events: List<Event>,
poison: Set<String>,
): Triple<Writer, Int, List<Pair<Event, Throwable>>> {
val writer = Writer(poison)
var accepted = 0
val poisoned = mutableListOf<Pair<Event, Throwable>>()
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<CancellationException> {
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())
}
}
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/HostStrikesTest.kt
+176
View File
@@ -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")
}
}
quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/Unreachability.kt
+48
View File
@@ -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
}
}
quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip66RelayMonitor/reachability/UnreachabilityTest.kt
+70
View File
@@ -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")))
}
}