feat(relayBench): deep resumable --download for million-event corpora

The corpus downloader previously sampled ~100k events max (40-page cap
per kind bucket) and held everything in memory with no failure recovery.
Rework it for full-depth timeline pulls:

- page the latest events newest-first with an inclusive until cursor
  (id-dedup absorbs the same-second overlap) instead of kind buckets
- no page cap: keep paging until the --limit target is met
- stream every unique event to an on-disk NDJSON spill instead of RAM
- checkpoint the pagination cursor per relay; interrupted downloads
  resume where they left off
- reconnect with exponential backoff on socket drops/timeouts
- filter deterministically droppable events (kind-5 deletions are ~40%
  of a live firehose, ephemerals, oversize) at page time so they never
  count toward the download goal

Verified with a 1M-event pull from relay.damus.io (~2.1 GB raw,
4100 pages, ~22 min through a proxy) feeding a full geode vs strfry
run; corpus prepared to exactly 1,000,000 events, fingerprint
141e746599d901f5.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012EZeWww5TJnzBZKPoc6mvU

relayBench/README.md

@@ -60,7 +60,7 @@ two runs are comparable only when fingerprints match.
 | **real dump** | `--real` | The `quartz` test fixture `nostr_vitor_startup_data.json.gz` — ~31k unique real events from 2024 with a rich kind mix (notes, chats, DMs, zaps, reports, communities). |
 | **contact lists** | `--corpus contact-lists.gz --limit 100000 --max-event-bytes 1048576 --max-tags 20000` | 2.1M real kind-3 contact lists (heavy events, ~1.3 kB avg, up to 100+ kB). Grab it with `pip install gdown && gdown 1yyC93xY9sDsEsa351ZAMhtAXwBUh3LYT`. Raising the size/tag caps reconfigures strfry to match, so both relays still accept the full stream. |
 | **any dump** | `--corpus FILE` | NDJSON or a single JSON array, gzipped or plain (sniffed by magic bytes). |
-| **fresh download** | `--download [urls]` | Pages recent events out of public relays (damus/nos.lol/primal by default). |
+| **fresh download** | `--download [urls]` | Pages the latest events out of public relays newest-first (damus/nos.lol/primal by default), `--limit N` deep — built for million-event pulls: streams to an on-disk spill, checkpoints the pagination cursor, resumes interrupted downloads, reconnects on drops. E.g. `--download wss://relay.damus.io --limit 1000000`. |
 
 Every source goes through the same preparation: dedup by id, drop unsigned
 events (NIP-17 rumors), kind-5 deletions and ephemerals (order-dependent or

relayBench/src/main/kotlin/com/vitorpamplona/relaybench/Main.kt

@@ -231,6 +231,7 @@ private fun loadCorpus(
                 options.cacheDir,
                 http,
                 log,
+                options.maxEventBytes,
             )
         else ->
             CorpusSource.synthetic(

relayBench/src/main/kotlin/com/vitorpamplona/relaybench/corpus/CorpusDownloader.kt

@@ -21,28 +21,42 @@
 package com.vitorpamplona.relaybench.corpus
 
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
-import com.vitorpamplona.quartz.nip01Core.core.Event
-import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.relaybench.NostrSocket
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeoutOrNull
 import okhttp3.OkHttpClient
 import java.io.File
+import java.io.FileWriter
+import java.io.Writer
+import java.util.concurrent.TimeUnit
 
 /**
- * Assembles a fresh real-world corpus by paginating recent events out of
- * public relays. Complements the checked-in dump: use this when the corpus
- * should reflect *today's* event mix. The result goes through the same
- * [CorpusSource.prepare] pipeline (dedup, verify, filter) as every other
- * source, then is cached as NDJSON.
+ * Assembles a fresh real-world corpus by paginating the *latest* events out
+ * of public relays — newest first, no kind filter, `until`-paged all the way
+ * down until the target is met. Complements the checked-in dump: use this
+ * when the corpus should reflect today's live event mix.
+ *
+ * Built to survive million-event pulls over flaky links:
+ *  - every unique event is appended to an on-disk NDJSON spill as it
+ *    arrives, so nothing is held in memory during the download;
+ *  - pagination progress (`until` cursor per relay) is checkpointed next to
+ *    the spill, so a crashed or interrupted run resumes where it left off;
+ *  - the socket is reconnected with backoff on timeouts/drops.
+ *
+ * The finished spill goes through the same [CorpusSource.prepare] pipeline
+ * (dedup, verify, filter, sort) as every other source, then is cached as
+ * NDJSON + manifest.
  */
 object CorpusDownloader {
     val DEFAULT_RELAYS = listOf("wss://relay.damus.io", "wss://nos.lol", "wss://relay.primal.net")
 
-    private val KIND_BUCKETS = listOf(listOf(0, 3), listOf(1), listOf(6, 7), listOf(9735), listOf(30023, 1111))
+    /** Stay inside the strictest common relay page cap (damus: max_limit 500). */
+    private const val PAGE_LIMIT = 500
+    private const val PAGE_TIMEOUT_MS = 45_000L
+    private const val MAX_CONSECUTIVE_FAILURES = 10
+
     private val mapper = jacksonObjectMapper()
 
     fun download(
@@ -51,82 +65,224 @@ object CorpusDownloader {
         cacheDir: File,
         http: OkHttpClient,
         log: (String) -> Unit,
+        maxEventBytes: Int = CorpusSource.DEFAULT_MAX_EVENT_BYTES,
     ): Corpus {
-        val key = "corpus-download-${relayUrls.hashCode()}-n$target.ndjson"
-        val cached = File(cacheDir, key)
+        val tag = relayUrls.joinToString("+") { hostOf(it) }
+        val key = "corpus-download-$tag-n$target"
+        val cached = File(cacheDir, "$key.ndjson")
         if (cached.exists()) {
             log("  reusing downloaded corpus ${cached.name}")
             return CorpusIO.read(cached, source = "download:${relayUrls.joinToString(",")}")
         }
 
-        val perRelay = (target * 2 / relayUrls.size).coerceAtLeast(500)
-        val collected = LinkedHashMap<String, Event>(target * 2)
-        runBlocking {
-            relayUrls
-                .map { url ->
-                    async {
-                        runCatching { downloadFrom(url, perRelay, log) }
-                            .onFailure { log("  ! $url failed: ${it.message}") }
-                            .getOrDefault(emptyList())
-                    }
-                }.awaitAll()
-        }.flatten().forEach { collected.putIfAbsent(it.id, it) }
+        cacheDir.mkdirs()
+        val spill = File(cacheDir, "$key.raw.ndjson")
+        val checkpoint = File(cacheDir, "$key.progress.json")
 
-        log("  downloaded ${collected.size} unique events from ${relayUrls.size} relays")
-        val corpus = CorpusSource.prepare(collected.values.toList(), target, "download:${relayUrls.joinToString(",")}", log)
+        // Deterministically droppable events (kind-5 deletions — ~40% of a
+        // live firehose! — ephemerals, oversize) are filtered at page time
+        // and never count toward the goal, so only a small headroom is left
+        // for what preparation alone can catch (invalid sigs, canonical-size
+        // edge cases, cross-relay duplicates).
+        val needed = if (relayUrls.size == 1) target + target / 25 else target * 2
+
+        val ids = HashSet<String>(needed * 2)
+        val cursors = HashMap<String, Long>()
+        if (spill.exists()) {
+            spill.useLines { lines ->
+                lines.filter { it.isNotBlank() }.forEach { line ->
+                    runCatching { mapper.readTree(line) }
+                        .getOrNull()
+                        ?.get("id")
+                        ?.asText()
+                        ?.let { ids.add(it) }
+                }
+            }
+            if (checkpoint.exists()) {
+                runCatching { mapper.readTree(checkpoint.readText()) }.getOrNull()?.fields()?.forEach { (url, until) ->
+                    cursors[url] = until.asLong()
+                }
+            }
+            log("  resuming download: ${ids.size} events already spilled to ${spill.name}")
+        }
+
+        val ws =
+            http
+                .newBuilder()
+                .pingInterval(15, TimeUnit.SECONDS)
+                .readTimeout(0, TimeUnit.MILLISECONDS)
+                .build()
+
+        spill.appendText("") // ensure the file exists even if the relay yields nothing
+        FileWriter(spill, true).buffered(1 shl 16).use { out ->
+            runBlocking {
+                // Even split across relays (the last one may top up the rest)
+                // so a multi-relay corpus actually mixes sources.
+                val perRelay = (needed + relayUrls.size - 1) / relayUrls.size
+                relayUrls.forEachIndexed { i, url ->
+                    val goal = if (i == relayUrls.lastIndex) needed else minOf(needed, ids.size + perRelay)
+                    if (ids.size < goal) {
+                        downloadFrom(url, goal, ids, out, cursors, checkpoint, ws, maxEventBytes, log)
+                    }
+                }
+            }
+        }
+
+        log("  downloaded ${ids.size} unique events from ${relayUrls.size} relay(s)")
+        if (ids.size < needed) {
+            log("  ! relays exhausted below the $needed-event goal — preparing what we have")
+        }
+
+        log("  preparing (dedup, filter, verify signatures)…")
+        val raw = CorpusIO.read(spill).events
+        val corpus = CorpusSource.prepare(raw, target, "download:${relayUrls.joinToString(",")}", log)
         CorpusIO.write(cached, corpus)
+        checkpoint.delete()
+        log("  cached prepared corpus to ${cached.path}")
         return corpus
     }
 
+    private fun hostOf(url: String) =
+        url
+            .removePrefix("wss://")
+            .removePrefix("ws://")
+            .trimEnd('/')
+            .replace(Regex("[^A-Za-z0-9.-]"), "_")
+
+    /**
+     * Pages `{"limit":500,"until":T}` down the relay's timeline, appending
+     * unique events to [out]. The `until` cursor stays *inclusive* of the
+     * oldest seen second (id-dedup absorbs the overlap) so events sharing a
+     * created_at are not skipped; only a full page of nothing-new at the same
+     * cursor — >500 events in one second — forces the cursor past it.
+     */
     private suspend fun downloadFrom(
         url: String,
-        target: Int,
+        goal: Int,
+        ids: MutableSet<String>,
+        out: Writer,
+        cursors: MutableMap<String, Long>,
+        checkpoint: File,
+        http: OkHttpClient,
+        maxEventBytes: Int,
         log: (String) -> Unit,
-    ): List<Event> {
-        val http = OkHttpClient.Builder().build()
-        val socket = NostrSocket.connect(http, url)
-        val events = ArrayList<Event>(target)
+    ) {
+        var until: Long? = cursors[url]
+        var socket: NostrSocket? = null
+        var failures = 0
+        var pages = 0
+        var added = 0
+        val startedAt = System.nanoTime()
+
+        fun saveCheckpoint() {
+            until?.let { cursors[url] = it }
+            runCatching { checkpoint.writeText(mapper.writeValueAsString(cursors)) }
+        }
+
         try {
-            for (kinds in KIND_BUCKETS) {
-                var until: Long? = null
-                var pages = 0
-                val quota = target / KIND_BUCKETS.size
-                var got = 0
-                while (got < quota && pages < 40) {
-                    val filter = Filter(kinds = kinds, limit = 500, until = until)
-                    val page = requestPage(socket, "dl-${kinds.first()}-$pages", filter) ?: break
-                    if (page.isEmpty()) break
-                    events += page
-                    got += page.size
-                    until = page.minOf { it.createdAt } - 1
-                    pages++
+            while (ids.size < goal) {
+                val sock =
+                    socket ?: runCatching { NostrSocket.connect(http, url) }.getOrElse {
+                        failures++
+                        if (failures > MAX_CONSECUTIVE_FAILURES) {
+                            log("  ! $url: giving up after $failures consecutive failures (${it.message})")
+                            return
+                        }
+                        delay(minOf(1000L shl (failures - 1), 30_000L))
+                        null
+                    } ?: continue
+                socket = sock
+
+                val filter = Filter(limit = PAGE_LIMIT, until = until)
+                val page = requestPage(sock, "dl-$pages", filter)
+                if (page == null) {
+                    // Timeout or dead socket: reconnect and retry the same cursor.
+                    runCatching { sock.disconnect() }
+                    socket = null
+                    failures++
+                    if (failures > MAX_CONSECUTIVE_FAILURES) {
+                        log("  ! $url: giving up after $failures consecutive page failures")
+                        return
+                    }
+                    delay(minOf(1000L shl (failures - 1), 30_000L))
+                    continue
+                }
+                failures = 0
+                pages++
+
+                if (page.isEmpty()) break // end of the relay's timeline
+
+                var fresh = 0
+                for (event in page) {
+                    // Skip what preparation would deterministically drop, so
+                    // they never count toward the goal: deletions, ephemerals
+                    // (never queryable), events over the size cap.
+                    if (event.kind == 5 || event.kind in 20000..29999) continue
+                    if (event.json.length > maxEventBytes) continue
+                    if (ids.add(event.id)) {
+                        out.write(event.json)
+                        out.write("\n")
+                        fresh++
+                    }
+                }
+                added += fresh
+                val oldest = page.minOf { it.createdAt }
+                until =
+                    if (fresh == 0 && until != null && oldest >= until!!) {
+                        // >PAGE_LIMIT events in this second and we have them all.
+                        until!! - 1
+                    } else {
+                        oldest
+                    }
+
+                out.flush()
+                saveCheckpoint()
+
+                if (page.size < PAGE_LIMIT && fresh == 0) break // exhausted
+
+                if (pages % 50 == 0) {
+                    val elapsed = (System.nanoTime() - startedAt) / 1e9
+                    val rate = added / elapsed
+                    val remaining = (goal - ids.size).coerceAtLeast(0)
+                    val eta = if (rate > 0) (remaining / rate).toLong() else -1
+                    log(
+                        "  $url: ${ids.size} events (${"%.0f".format(rate)}/s, page $pages, " +
+                            "cursor $until, eta ${if (eta >= 0) "${eta / 60}m${eta % 60}s" else "?"})",
+                    )
                 }
             }
         } finally {
-            socket.disconnect()
-            http.dispatcher.executorService.shutdown()
+            runCatching { socket?.disconnect() }
         }
-        log("  $url: ${events.size} events")
-        return events
+        log("  $url: $added events in $pages pages")
     }
 
-    /** One REQ page; null on timeout or socket close. */
+    private class PagedEvent(
+        val id: String,
+        val kind: Int,
+        val createdAt: Long,
+        val json: String,
+    )
+
+    /** One REQ page; null on timeout or socket close (caller reconnects). */
     private suspend fun requestPage(
         socket: NostrSocket,
         subId: String,
         filter: Filter,
-    ): List<Event>? =
-        withTimeoutOrNull(30_000) {
-            socket.req(subId, filter.toJson())
-            val page = ArrayList<Event>(filter.limit ?: 500)
+    ): List<PagedEvent>? =
+        withTimeoutOrNull(PAGE_TIMEOUT_MS) {
+            if (!socket.req(subId, filter.toJson())) return@withTimeoutOrNull null
+            val page = ArrayList<PagedEvent>(filter.limit ?: PAGE_LIMIT)
             for (raw in socket.incoming) {
                 val node = runCatching { mapper.readTree(raw) }.getOrNull() ?: continue
                 when (node[0]?.asText()) {
                     "EVENT" ->
                         if (node[1]?.asText() == subId) {
-                            runCatching { OptimizedJsonMapper.fromJson(node[2].toString()) }
-                                .getOrNull()
-                                ?.let { page.add(it) }
+                            val event = node[2] ?: continue
+                            val id = event["id"]?.asText() ?: continue
+                            val kind = event["kind"]?.asInt() ?: continue
+                            val createdAt = event["created_at"]?.asLong() ?: continue
+                            page.add(PagedEvent(id, kind, createdAt, event.toString()))
                         }
                     "EOSE", "CLOSED" ->
                         if (node[1]?.asText() == subId) {
@@ -135,6 +291,6 @@ object CorpusDownloader {
                         }
                 }
             }
-            page
+            null // channel closed without EOSE — socket died mid-page
         }
 }