mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-12 01:07:46 +00:00
SyncCoverage: one band interval cannot speak for several kinds
A band held ONE created_at interval per (relay, filter). For a filter naming several kinds that is a claim no walk can support: ask for `kinds: [0, 30382]`, find profiles going back years and score cards only from last month, and the band records 2020..now for the pair. The next run then skips that whole interior for BOTH — so score cards written inside it are never asked for again, and nothing anywhere says so. A long-lived kind vouched for a short-lived one. Band.spans is now per kind. Each carries only the evidence actually collected for it, so the profile kind keeps its wide interval and the score kind keeps its narrow one, and legs() re-opens the interior for the second while still skipping it for the first. Three things keep the cost of that where it was: - legs() REGROUPS kinds by the windows they want. Identical coverage — the common case, and the only case until they diverge — collapses back into one ask, so a filter that produced two legs still produces two rather than two per kind. Only a kind whose evidence genuinely differs earns its own. - A finished reconcile needs no per-kind evidence and is given none: negentropy compares the filter's whole id set in one pass, so it covers every kind in the filter or none. Only the PAGED path changed. - Filters naming no kinds keep a single span under ALL_KINDS, which is the same claim as before, correctly scoped to the case where it is the only claim available. record() takes observedByKind, and SyncCoverage.observe() accumulates it as events arrive — replacing the pair of hand-rolled vars each caller kept, and moving the per-event isPlausible guard in with it. A paged walk over a MULTI-kind filter that supplies none earns no band at all, loudly, once: attributing one interval to every kind is exactly the over-claim this removes, and a band that over-claims skips events silently, which is worse than re-reading them. Single-kind filters are untouched — there the aggregate always was the per-kind answer. The state file gains a per-kind `spans` object and keeps `min`/`max` as the outer edges, so a rollback to a binary from before this reads the file and behaves as it always did. A file written BEFORE this loads its one interval under ALL_KINDS — the old, wider claim, kept rather than discarded because discarding it would re-download every upstream's corpus once on upgrade. The first per-kind walk replaces it. All 26 existing SyncCoverage tests pass unchanged, which is the evidence that single-kind behaviour did not move. The five new ones were checked against the pre-fix rule reinstated in place: the two behavioural ones fail there and pass here. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Vitor Pamplona
co-authored by
Claude Opus 5
Claude Opus 5
parent
e822911d09
commit
42a91ffb79
@@ -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<Int, SyncCoverage.Span>()
|
||||
|
||||
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
|
||||
|
||||
@@ -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<Int, SyncCoverage.Span> {
|
||||
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)
|
||||
},
|
||||
)
|
||||
}
|
||||
},
|
||||
)
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user