diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt index c98e050e37..4d1411593b 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/MirrorWorker.kt @@ -498,6 +498,12 @@ class MirrorWorker( val syncStartedAt = TimeUtils.now() var seenMin: Long? = null var seenMax: Long? = null + // Per KIND as well as in aggregate: one interval for a + // multi-kind filter lets a long-lived kind vouch for a + // short-lived one, and the band then skips the interior for + // both. The aggregate is still tracked because the reconcile + // path records against the leg's floor, not per kind. + val seenByKind = mutableMapOf() fun observe(event: Event) { // Same containment as the live path: even a trusted @@ -509,6 +515,7 @@ class MirrorWorker( seenMin = minOf(seenMin ?: event.createdAt, event.createdAt) seenMax = maxOf(seenMax ?: event.createdAt, event.createdAt) } + SyncCoverage.observe(seenByKind, event.kind, event.createdAt) handoff.trySendBlocking(event) } else { filtered.incrementAndGet() @@ -537,6 +544,7 @@ class MirrorWorker( } catch (e: NegentropySyncException) { seenMin = null seenMax = null + seenByKind.clear() // The watchdog matches negentropySync's default rather // than fetchAllPages' shorter one: a paged catch-up // sits behind the same slow upstreams. @@ -558,6 +566,13 @@ class MirrorWorker( seenMin, seenMax?.coerceAtMost(syncStartedAt), paged = true, + // Capped the same way the aggregate is: one + // future-dated event must not lift a kind's ceiling + // past what was actually asked for. + observedByKind = + seenByKind.mapValues { (_, span) -> + SyncCoverage.Span(span.min, span.max.coerceAtMost(syncStartedAt)) + }, ) } else { val legFloor = leg.since ?: initialSince diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt index b31c51a60d..58b0b23310 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/mirror/SyncCoverageFile.kt @@ -100,8 +100,7 @@ class SyncCoverageFile( root.mapValues { (_, v) -> val o = v.jsonObject SyncCoverage.Band( - o.getValue("min").jsonPrimitive.long, - o.getValue("max").jsonPrimitive.long, + spansOf(o), o["complete"]?.jsonPrimitive?.boolean ?: false, o["fullAt"]?.jsonPrimitive?.long ?: 0L, ) @@ -112,6 +111,30 @@ class SyncCoverageFile( } } + /** + * The per-kind spans, or the single pre-split span read as covering every + * kind under [SyncCoverage.ALL_KINDS]. + * + * A file written before coverage was tracked per kind carries only + * `min`/`max`, and that is exactly the over-wide claim per-kind spans + * exist to stop — so it is loaded as what it always meant rather than + * discarded, and the first paged walk that reports per kind replaces it. + * Dropping it instead would re-download every upstream's corpus once on + * upgrade, which is the cost bands exist to avoid. + */ + private fun spansOf(o: JsonObject): Map { + o["spans"]?.jsonObject?.let { spans -> + return spans.entries.associate { (kind, v) -> + val span = v.jsonObject + kind.toInt() to SyncCoverage.Span(span.getValue("min").jsonPrimitive.long, span.getValue("max").jsonPrimitive.long) + } + } + return mapOf( + SyncCoverage.ALL_KINDS to + SyncCoverage.Span(o.getValue("min").jsonPrimitive.long, o.getValue("max").jsonPrimitive.long), + ) + } + @Synchronized private fun save() { runCatching { @@ -121,10 +144,30 @@ class SyncCoverageFile( put( key, buildJsonObject { + // min/max are the outer edges across every + // kind, and are written for two readers: a + // human debugging why an upstream re-synced, + // and a ROLLBACK — a binary from before spans + // were per kind reads these and behaves as it + // always did, rather than failing to parse. put("min", band.minCreatedAt) put("max", band.maxCreatedAt) put("complete", band.complete) put("fullAt", band.fullAt) + put( + "spans", + buildJsonObject { + band.spans.forEach { (kind, span) -> + put( + kind.toString(), + buildJsonObject { + put("min", span.min) + put("max", span.max) + }, + ) + } + }, + ) }, ) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt index 35ba9b330a..4e6de68696 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverage.kt @@ -22,6 +22,7 @@ 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.Log import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.concurrent.ConcurrentMap @@ -64,23 +65,55 @@ class SyncCoverage( private val now: () -> Long = { TimeUtils.now() }, private val onChange: () -> Unit = {}, ) { + /** A covered `created_at` interval, inclusive at both ends. */ + data class Span( + val min: Long, + val max: Long, + ) { + fun widen(other: Span) = Span(minOf(min, other.min), maxOf(max, other.max)) + } + /** * What is already covered for one (relay, filter) pair. * + * [spans] is PER KIND, and that is the whole point of it. A band used to + * hold one interval for the entire filter, which is a claim no multi-kind + * walk can support: ask for `kinds: [0, 30382]`, see profiles back to 2020 + * and score cards only from 2025, and the band reads 2020..2026 — so the + * next run skips 2020..2025 for BOTH, and the score cards in that interior + * are never asked for again. A long-lived kind vouched for a short-lived + * one. Per kind, each carries only the evidence actually collected for it. + * + * Filters that name no kinds at all cannot be split, so they keep a single + * span under [ALL_KINDS] — the same claim as before, correctly scoped to + * the case where it is the only claim available. + * * [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. + * skip its older leg. It is a property of the BAND rather than of a span: + * a reconcile compares the filter's whole id set at once, so it either + * covers every kind in it or none. * * [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 spans: Map, val complete: Boolean = false, val fullAt: Long = 0, - ) + ) { + /** The outer edges across every kind — for logging and for the file's compatibility fields. */ + val minCreatedAt: Long get() = spans.values.minOfOrNull { it.min } ?: 0 + val maxCreatedAt: Long get() = spans.values.maxOfOrNull { it.max } ?: 0 + + /** Widen each kind by its counterpart, keeping kinds only one side knows. */ + fun widen(other: Band): Band { + val merged = spans.toMutableMap() + for ((kind, span) in other.spans) merged[kind] = merged[kind]?.widen(span) ?: span + return Band(merged, complete || other.complete, fullAt) + } + } private val bands = ConcurrentMap() @@ -114,31 +147,68 @@ class SyncCoverage( // 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() + if (band.spans.isEmpty()) return listOf(filter) - // Older: up to and including the band's floor, but not past the - // filter's (or, when the filter has no `since`, the caller's - // [floor] — a sync window the filter itself must not carry, or it - // would change the band's key every run). A complete band compared - // its whole range already, but only down to the floor it ran - // against: a caller now reaching deeper — a raised backfill window - // — re-opens the span below the band. + val kinds = filter.kinds + if (kinds.isNullOrEmpty()) { + // Nothing to split by. One span, exactly as before. + return windows(filter, band.spans[ALL_KINDS], band.complete, floor) + .map { (since, until) -> filter.copy(since = since, until = until) } + } + + // Per kind, then REGROUPED by the windows each one wants. Kinds whose + // coverage agrees — the overwhelmingly common case, and the only case + // at all until they diverge — collapse back into one ask, so a filter + // that used to produce two legs still produces two rather than two per + // kind. Only a kind whose evidence genuinely differs earns its own. + val byWindows = LinkedHashMap>, MutableList>() + for (kind in kinds) { + // ALL_KINDS as the fallback: a band written before coverage was + // tracked per kind, restored from such a file. It carries the old, + // wider claim for every kind — the behaviour this replaces — and + // self-corrects on the first paged walk that reports per kind. + val span = band.spans[kind] ?: band.spans[ALL_KINDS] + byWindows.getOrPut(windows(filter, span, band.complete, floor)) { mutableListOf() }.add(kind) + } + return byWindows.flatMap { (windows, group) -> + windows.map { (since, until) -> filter.copy(kinds = group, since = since, until = until) } + } + } + + /** + * The `(since, until)` pairs still outstanding for ONE span — the leg + * arithmetic, with the filter's own bounds applied and nothing else. + * A null [span] means no evidence at all, so the whole filter is wanted. + */ + private fun windows( + filter: Filter, + span: Span?, + complete: Boolean, + floor: Long?, + ): List> { + if (span == null) return listOf(filter.since to filter.until) + val out = mutableListOf>() + + // Older: up to and including the span's floor, but not past the + // filter's (or, when the filter has no `since`, the caller's [floor] — + // a sync window the filter itself must not carry, or it would change + // the band's key every run). A complete band compared its whole range + // already, but only down to the floor it ran against: a caller now + // reaching deeper — a raised backfill window — re-opens the span below. val since = filter.since ?: floor val wantsOlder = - if (band.complete) { - since != null && since < band.minCreatedAt + if (complete) { + since != null && since < span.min } else { - since == null || band.minCreatedAt >= since + since == null || span.min >= since } - if (wantsOlder) { - legs.add(filter.copy(until = minOf(band.minCreatedAt, filter.until ?: Long.MAX_VALUE))) - } + if (wantsOlder) out.add(filter.since to minOf(span.min, 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))) + // Newer: from the span's ceiling on, but not past the filter's. + if (filter.until == null || span.max <= filter.until) { + out.add(maxOf(span.max, filter.since ?: Long.MIN_VALUE) to filter.until) } - return legs + return out } /** @@ -162,21 +232,59 @@ class SyncCoverage( observedMax: Long?, paged: Boolean, reconciledThrough: Long? = null, + observedByKind: Map? = null, ) { if (reconciledThrough != null) { - put(url, filter, observedMin ?: reconciledThrough, reconciledThrough, complete = true) + // A reconcile compares the filter's whole id set in one pass, so + // the span it earns is the same for every kind the filter names — + // no per-kind evidence needed or possible. + val span = Span(observedMin ?: reconciledThrough, reconciledThrough) + put(url, filter, kindsOf(filter).associateWith { span }, complete = true) return } if (!paged) return + + if (observedByKind != null) { + // Guarded per span for the same reason the aggregate is below. + val plausible = + observedByKind.filterValues { + isPlausible(it.min, now()) && isPlausible(it.max, now()) + } + if (plausible.isEmpty()) return + put(url, filter, plausible, complete = false) + return + } + + // No per-kind evidence. For a filter naming one kind (or none) the + // aggregate IS the per-kind answer and nothing is lost. For a filter + // naming several it is not: attributing one interval to all of them is + // exactly the over-claim [Band.spans] exists to stop, and a band that + // over-claims skips events silently — strictly worse than re-reading + // them. So record nothing and say why, once. The caller resumes as if + // it had no band, which is where it was before bands existed. + val kinds = kindsOf(filter) + if (kinds.size > 1) { + if (!warnedAboutUnattributed) { + warnedAboutUnattributed = true + Log.w("SyncCoverage") { + "paged record for a ${kinds.size}-kind filter with no per-kind spans — no band recorded, so this " + + "walk will not resume. Pass observedByKind (see SyncCoverage.observe) to earn one." + } + } + 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) + put(url, filter, kinds.associateWith { Span(observedMin, observedMax) }, complete = false) } + /** The kinds a band is keyed by: the filter's, or [ALL_KINDS] when it names none. */ + private fun kindsOf(filter: Filter): List = filter.kinds?.takeIf { it.isNotEmpty() } ?: listOf(ALL_KINDS) + /** * 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 @@ -185,22 +293,12 @@ class SyncCoverage( private fun put( url: NormalizedRelayUrl, filter: Filter, - min: Long, - max: Long, + spans: Map, complete: Boolean, ) { - val fresh = Band(min, max, complete, now()) + val fresh = Band(spans, 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, - ) - } + if (isStale(old)) new else old.widen(new) } onChange() } @@ -276,7 +374,45 @@ class SyncCoverage( return "${url.url} $fingerprint" } + // One line per process, not per walk: the point is to tell a caller it has + // not been migrated, and repeating it every leg would bury the log it is + // trying to be read in. + private var warnedAboutUnattributed = false + companion object { + /** + * The span key for a filter that names no kinds, and the fallback for + * a band restored from a file written before spans were per kind. + * Negative because NIP-01 kinds are not. + */ + const val ALL_KINDS = -1 + + /** + * Widen [into] with one event's stamp, so a caller can accumulate the + * per-kind evidence [record] wants as events arrive: + * + * val seen = mutableMapOf() + * ... onEvent { SyncCoverage.observe(seen, it.kind, it.createdAt) } + * coverage.record(url, filter, …, paged = true, observedByKind = seen) + * + * Implausible stamps are dropped here rather than by each caller — + * per EVENT, never over a leg's aggregate, because one misdated event + * among hundreds of thousands would otherwise discard the whole band. + * + * Not synchronized: it replaces a pair of plain `var`s at each call + * site and is meant for the same single-consumer callback. + */ + fun observe( + into: MutableMap, + kind: Int, + createdAt: Long, + now: Long = TimeUtils.now(), + ) { + if (!isPlausible(createdAt, now)) return + val one = Span(createdAt, createdAt) + into[kind] = into[kind]?.widen(one) ?: one + } + // More filter instances than any deliberate configuration holds; only // a caller rebuilding filters per cycle ever reaches it. private const val MAX_FINGERPRINTS = 1_000 diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt index 53120624a4..87cac2c977 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/SyncCoverageTest.kt @@ -382,4 +382,124 @@ class SyncCoverageTest { 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") } + + // ---- per-kind spans: one interval cannot speak for several kinds ------- + + private val mixed = Filter(kinds = listOf(0, 30382)) + + /** Does any leg still ask [kind] about the instant [at]? */ + private fun reaches( + legs: List, + kind: Int, + at: Long, + ) = legs.any { + (it.kinds?.contains(kind) ?: true) && + (it.since ?: Long.MIN_VALUE) <= at && + at <= (it.until ?: Long.MAX_VALUE) + } + + @Test + fun `a long-lived kind no longer vouches for a short-lived one`() { + // THE BUG. Ask for profiles and score cards together: the relay has + // profiles going back years and score cards only from last month. One + // interval per band recorded 2020..now for the pair, and the next run + // skipped that whole interior for BOTH — so score cards written inside + // it were never asked for again, and nothing anywhere said so. + val c = SyncCoverage() + c.record( + relay, + mixed, + null, + null, + paged = true, + observedByKind = + mapOf( + 0 to SyncCoverage.Span(1_600_000_000L, 1_700_000_000L), + 30382 to SyncCoverage.Span(1_690_000_000L, 1_700_000_000L), + ), + ) + val legs = c.legs(relay, mixed) + + assertTrue(!reaches(legs, 0, 1_650_000_000L), "kind 0 really was walked there — do not re-read it") + assertTrue(reaches(legs, 30382, 1_650_000_000L), "kind 30382 never was, and must still be asked") + // Both keep the ground they actually earned. + assertTrue(!reaches(legs, 30382, 1_695_000_000L), "…but not its own covered interior") + assertTrue(reaches(legs, 0, 1_500_000_000L), "and both still reach below everything walked") + } + + @Test + fun `kinds whose coverage agrees stay a single ask`() { + // The cost control. Splitting per kind would turn two legs into two + // per kind on every filter, which is the common case made worse to fix + // the rare one. Kinds are regrouped by the windows they want, so + // identical coverage collapses back to exactly what it was before. + val c = SyncCoverage() + val span = SyncCoverage.Span(1_690_000_000L, 1_700_000_000L) + c.record(relay, mixed, null, null, paged = true, observedByKind = mapOf(0 to span, 30382 to span)) + + val legs = c.legs(relay, mixed) + assertEquals(2, legs.size, "two legs, not two per kind") + assertEquals(listOf(0, 30382), legs[0].kinds, "and both kinds ride in one ask") + } + + @Test + fun `a multi-kind paged walk with no per-kind evidence earns no band`() { + // The caller did not say which kind it saw where, so the only band + // available is the over-wide one. Refused: a band that over-claims + // skips events silently, which is worse than re-reading them. The + // walk resumes from nothing, exactly as it did before bands existed. + val c = SyncCoverage() + c.record(relay, mixed, 1_690_000_000L, 1_700_000_000L, paged = true) + + assertNull(c.band(relay, mixed)) + assertEquals(listOf(mixed), c.legs(relay, mixed)) + + // A filter naming ONE kind is unaffected: there, the aggregate IS the + // per-kind answer and nothing was ever ambiguous about it. + c.record(relay, profiles, 1_690_000_000L, 1_700_000_000L, paged = true) + assertEquals(2, c.legs(relay, profiles).size) + } + + @Test + fun `a finished reconcile covers every kind the filter names`() { + // Negentropy compares the filter's whole id set in one pass, so it + // either covers every kind in it or none — no per-kind evidence needed, + // and none invented. + val c = SyncCoverage() + c.record(relay, mixed, null, null, paged = false, reconciledThrough = 1_700_000_000L) + + assertEquals(setOf(0, 30382), c.band(relay, mixed)!!.spans.keys) + val legs = c.legs(relay, mixed) + assertEquals(1, legs.size, "complete: no older leg, and one shared newer one") + assertEquals(1_700_000_000L, legs[0].since) + } + + @Test + fun `a band restored from a pre-split file still narrows every kind`() { + // Files written before spans were per kind carry one interval. It is + // the old, wider claim — loaded as what it always meant rather than + // discarded, because discarding it would re-download every upstream's + // corpus once on upgrade. The first per-kind walk replaces it. + val seed = SyncCoverage() + // A plausible span, or record() correctly drops it and there is no key to read. + seed.record(relay, mixed, null, null, paged = true, observedByKind = mapOf(0 to SyncCoverage.Span(1_690_000_000L, 1_700_000_000L))) + val key = seed.export().keys.single() + + val restored = SyncCoverage() + restored.restore( + mapOf( + key to + SyncCoverage.Band( + mapOf(SyncCoverage.ALL_KINDS to SyncCoverage.Span(1_690_000_000L, 1_700_000_000L)), + complete = false, + fullAt = now(), + ), + ), + ) + + val legs = restored.legs(relay, mixed) + assertEquals(2, legs.size, "one shared pair of legs, which is the old behaviour exactly") + assertEquals(listOf(0, 30382), legs[0].kinds) + assertEquals(1_690_000_000L, legs[0].until) + } }