perf: drop per-event iterator alloc; harden count/fetchFirst drain loops

Audit follow-up on the timeout work.

- fetchAllPages matched each event with `for ((i, f) in activeFilters)`.
  That destructuring form allocates an Iterator on every event, on the
  relay's reader thread, for the whole download -- millions of short-lived
  objects in a bulk walk. Switched to an indexed loop, which is why quartz
  uses the fast* operators elsewhere in hot event paths (those only cover
  Array, so a List needs the index form).

- count() and fetchFirst() drain their channels in an inner loop that only
  suspends when the channel is empty, so a backlog was consumed with no
  cancellation check: neither the idle window expiring nor the caller
  giving up could interrupt it mid-drain. Added an explicit ensureActive(),
  matching the check fetchAllPages already does per page.

- count() could also lose a result that arrived after the last window
  closed but before unsubscribe, understating the returned map. Added the
  post-loop tryReceive drain that fetchAllWithHooks and fetchFirst have.

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

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientCountExt.kt

@@ -32,7 +32,9 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip45Count.HyperLogLog
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.coroutines.coroutineContext
 
 /**
  * Sends a NIP-45 COUNT query to a single relay and suspends until
@@ -146,6 +148,11 @@ suspend fun INostrClient.count(
             val progressed =
                 withTimeoutOrNull(idleTimeoutMs) {
                     while (true) {
+                        // Cancellation (this window expiring, or the caller giving up)
+                        // only lands at a suspension point, and receive() does not
+                        // suspend while the channel has buffered results — so check
+                        // explicitly rather than draining a backlog uninterruptibly.
+                        coroutineContext.ensureActive()
                         val (relay, result) = resultChannel.receive()
                         // put() returns the previous value: null means this relay
                         // had not answered yet, i.e. real progress.
@@ -155,6 +162,13 @@ suspend fun INostrClient.count(
                 }
             if (progressed == null) break
         }
+
+        // A result can land after the last window closed but before we unsubscribe;
+        // it costs nothing to keep, and dropping it would understate the count.
+        while (true) {
+            val (relay, result) = resultChannel.tryReceive().getOrNull() ?: break
+            results[relay] = result
+        }
     } finally {
         subIdToRelay.keys.forEach { unsubscribe(it) }
         removeConnectionListener(listener)

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt

@@ -201,7 +201,16 @@ suspend fun INostrClient.fetchAllPages(
                             // next page skip events a co-resident normal filter still needs.
                             var atLeastOne = false
                             var advancesCursor = false
-                            for ((index, filter) in activeFilters) {
+                            // Indexed loop, not `for ((i, f) in activeFilters)`: this runs for
+                            // EVERY event on the relay's reader thread (millions in a bulk
+                            // download) and the destructuring form allocates an Iterator per
+                            // event. Same reason quartz uses the `fast*` operators elsewhere
+                            // in hot event paths — those only cover Array, so a List needs
+                            // the index form.
+                            for (i in activeFilters.indices) {
+                                val active = activeFilters[i]
+                                val index = active.index
+                                val filter = active.value
                                 if (matchCountPerFilter[index] < (filter.limit ?: Int.MAX_VALUE) && filter.match(event)) {
                                     matchCountPerFilter[index]++
                                     atLeastOne = true

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchFirstExt.kt

@@ -29,8 +29,10 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.selects.select
 import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.coroutines.coroutineContext
 
 suspend fun INostrClient.fetchFirst(
     relay: String,
@@ -139,6 +141,12 @@ suspend fun INostrClient.fetchFirst(
             val progressed =
                 withTimeoutOrNull(idleTimeoutMs) {
                     while (true) {
+                        // Cancellation (this window expiring, or the caller giving up)
+                        // only lands at a suspension point, and select() completes
+                        // without suspending while either channel has something
+                        // buffered — so check explicitly rather than draining a
+                        // backlog of chatter uninterruptibly.
+                        coroutineContext.ensureActive()
                         val advanced =
                             select<Boolean> {
                                 eventChannel.onReceive { event ->