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

perf(graperank): adaptive idle-EOSE cutoff (--eose-idle-ms)

Browse Source 
Measured: relays deliver their events in ~0.6s then sit ~4.6s (86% of drain wall)
before sending EOSE — mostly relay-side (our pipeline adds only ~200ms). So instead
of waiting the full 10s fast window then parking, close a drain that has delivered
>=1 event and then gone silent for eoseIdleMs, treating it as complete ("eose-idle").

awaitTerminalOrQuiescent: the idle timer arms only AFTER the first event, so a relay
merely slow to answer still gets the full timeoutMs and is never cut prematurely; a
still-streaming relay keeps resetting the window. eose-idle paginates if the page was
capped and clears timeout strikes (it delivered), but joins notAnswered (no clean
EOSE, so its missing authors are retried elsewhere). Off by default (eoseIdleMs=0),
CLI --eose-idle-ms, so it can be A/B'd against the plain fast window.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MSW59hJtP4Yn8fnRUxc7F5
This commit is contained in:
Claude
2026-07-09 18:19:43 +00:00
parent a5c2a8b2c1
commit e02f00384a
2 changed files with 63 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/GrapeRankCommand.kt
+3
View File
@@ -464,6 +464,9 @@ object GrapeRankCommand {
insertBatchSize = args.intFlag("insert-batch", 500),
drainConcurrency = args.intFlag("drain-concurrency", 24),
timeoutEvictStrikes = args.intFlag("timeout-evict", 3),
// Adaptive EOSE cutoff: close a drain that delivered then fell silent
// this many ms, instead of waiting the full fast window. 0 = off.
eoseIdleMs = args.longFlag("eose-idle-ms", 0L),
// Cheap TCP reachability pre-probe (--no-probe to disable). No Tor
// transport here, so .onion relays are skipped on sight.
reachabilityProbe = if (args.bool("no-probe")) null else ::tcpReachable,
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/experimental/graperank/GrapeRankCrawler.kt
+60 -2
View File
@@ -208,6 +208,15 @@ class GrapeRankCrawler(
* permanently ignores an author's advertised home. See RelayReachabilityStore.
*/
val knownDeadRelays: Set<NormalizedRelayUrl> = emptySet(),
/**
* Adaptive EOSE cutoff, in ms; 0 disables (the plain fast window). Measured:
* relays deliver their events in ~0.6s then sit ~4.6s more before sending EOSE
* (86% of drain wall). When >0, a drain that has delivered ≥1 event and then
* gone silent for this long is treated as complete instead of waiting out the
* full [timeoutMs] and parking. The idle timer only arms AFTER the first event,
* so a relay merely slow to answer still gets the full [timeoutMs].
*/
val eoseIdleMs: Long = 0,
)
/** What the crawl fetched — the counters the caller reports and the graph is built from. */
@@ -960,6 +969,46 @@ class GrapeRankCrawler(
}
}
/**
* Fast-window wait with an adaptive EOSE cutoff. Measured: relays answer in
* ~0.6s then go quiet, but sit ~4.6s more before sending EOSE (86% of drain
* wall). So once a relay has delivered events AND then gone silent for [idleMs],
* treat it as complete ("eose-idle") instead of waiting out the full [hardCapMs]
* fast window and parking. Returns:
* - a terminal reason (eose/closed/cannot) if one arrives,
* - "eose-idle" if the relay delivered then fell quiet for [idleMs] (done),
* - null if [hardCapMs] elapsed while it was still streaming or never answered
* (→ park, exactly as the plain fast window would).
* The idle timer only ARMS after the first event, so a relay merely slow to send
* its first response gets the full [hardCapMs] and is never cut prematurely.
*/
private suspend fun awaitTerminalOrQuiescent(
done: CompletableDeferred<String>,
activity: Channel<Unit>,
idleMs: Long,
hardCapMs: Long,
): String? {
val start = TimeSource.Monotonic.markNow()
var sawEvent = false
while (true) {
val remaining = hardCapMs - start.elapsedNow().inWholeMilliseconds
if (remaining <= 0) return if (sawEvent) EOSE_IDLE else null
val window = if (sawEvent) minOf(idleMs, remaining) else remaining
val r =
withTimeoutOrNull(window) {
select {
done.onAwait { it }
activity.onReceive { ACTIVITY }
}
}
when (r) {
null -> return if (sawEvent) EOSE_IDLE else null // quiet after data → done; else park
ACTIVITY -> sawEvent = true // an event landed — arm/reset the idle window
else -> return r // terminal reason (eose/closed/cannot)
}
}
}
/**
* Fold one late-delivered event from a parked relay into the graph. Only the
* round loop calls this (directly or via [foldLateHarvest]), so graph state
@@ -1211,7 +1260,12 @@ class GrapeRankCrawler(
}
}
client.subscribe(subId, mapOf(subRelay to groupFilters), listener)
val reason = withTimeoutOrNull(config.timeoutMs) { done.await() }
val reason =
if (config.eoseIdleMs > 0) {
awaitTerminalOrQuiescent(done, activity, config.eoseIdleMs, config.timeoutMs)
} else {
withTimeoutOrNull(config.timeoutMs) { done.await() }
}
if (reason != null) {
// Terminal within the fast window — resolve this round.
val elapsedMs = mark.elapsedNow().inWholeMilliseconds
@@ -1235,7 +1289,9 @@ class GrapeRankCrawler(
val persisted = persist(drained)
// A full page from a clean EOSE may be the relay's cap, not the
// whole answer — background-paginate the remainder into lateHarvest.
if (reason == "eose") paginateIfCapped(subRelay, groupFilters, drained)
// Paginate on a clean EOSE and on an idle-cutoff: in both
// cases a full page may be the relay's cap, not the whole answer.
if (reason == "eose" || reason == EOSE_IDLE) paginateIfCapped(subRelay, groupFilters, drained)
// Alive if it EOSE'd or handed us anything; a connect-timeout that
// gave nothing (classifyDrainFailure leaves it retryable forever)
// earns a strike toward eviction instead.
@@ -1779,6 +1835,7 @@ class GrapeRankCrawler(
): Outcome =
when {
reason == "eose" -> if (parked) Outcome.SLOW_EOSE else Outcome.FAST_EOSE
reason == EOSE_IDLE -> Outcome.FAST_EOSE // delivered then quiet: a successful fast completion
reason == "timeout" -> if (parked) Outcome.PARK_TIMEOUT else Outcome.FAST_TIMEOUT
reason.startsWith("cannot") -> Outcome.CANNOT
reason.startsWith("closed:") -> {
@@ -1894,6 +1951,7 @@ class GrapeRankCrawler(
// (resets the window). A control string that can't collide with a relay's
// CLOSED/cannot message, which are the only other select results.
private const val ACTIVITY = "activity"
private const val EOSE_IDLE = " eose-idle"
// Times we re-query an unreachable user's outbox before giving up, so the
// crawl still terminates on a finite graph.