fix(quartz): audit fixes — VerifyAuthOnlyPolicy + small wins

Self-audit of the event-ingestion-batching changes turned up one
real bug + a handful of cleanups.

Fix: AUTH events skipped signature verification when
parallelVerify=true. Previous commit dropped VerifyPolicy from the
policy chain to avoid double-verifying EVENTs (the IngestQueue does
those off-thread). But VerifyPolicy.accept(AuthCmd) was the only
thing checking AUTH signatures — FullAuthPolicy verifies challenge
/ relay / expiry but trusts the sig. Removing VerifyPolicy let a
forged event mark a pubkey as authenticated.

Split VerifyPolicy into a parameterised base class
(VerifyEventsAndAuthPolicy) with two singletons:
- VerifyPolicy: verifies both EVENT and AUTH (existing default).
- VerifyAuthOnlyPolicy: verifies AUTH only — use when an
  IngestQueue does parallel EVENT verify, since AUTH commands
  bypass the queue entirely.

Geode's composePolicy now selects VerifyAuthOnlyPolicy when
parallelVerify is on, keeping AUTH signature checks intact while
still letting EVENT verify run in parallel on the writer's CPU
fan-out.

Cleanups (no behavior change):
- IngestQueue.processBatch was ~70 lines; split into verifyBatch /
  runInsertStage / dispatchOutcomes.
- Single-event verify shortcut: skip the coroutineScope + async
  dance for batch-of-1 (the common low-load case) and call the
  hook directly.
- Hoisted the "internal error: missing outcome" Rejected sentinel
  to a companion `missingOutcome` constant.
- Imported ClosedSendChannelException / ClosedReceiveChannelException
  instead of using the fully-qualified form inline.
- Dropped NostrServer's verifyEvent companion wrapper — direct
  lambda is just as cheap and the "single instance" comment was
  inaccurate.
- ObservableEventStore.batchInsert now uses requireNoNulls() rather
  than @Suppress("UNCHECKED_CAST"), since every index is provably
  populated.

geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt

@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.KindAllowDenyPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PubkeyAllowDenyPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RejectFutureEventsPolicy
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyAuthOnlyPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
@@ -104,10 +105,7 @@ fun main(args: Array<String>) {
     val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl)
 
     val policyBuilder: () -> IRelayPolicy = {
-        // When parallel verify is enabled the IngestQueue runs
+        composePolicy(config, advertisedUrl, requireAuth, verifySigs, parallelVerify)
-        // Schnorr verify off the WS pump, so the policy chain skips
-        // VerifyPolicy to avoid double-verifying every event.
-        composePolicy(config, advertisedUrl, requireAuth, verifySigs && !parallelVerify)
     }
 
     val stateFile = config.admin.state_file?.let { File(it) }
@@ -168,6 +166,7 @@ private fun composePolicy(
     advertisedUrl: com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl,
     requireAuth: Boolean,
     verifySigs: Boolean,
+    parallelVerify: Boolean,
 ): IRelayPolicy {
     val pieces = mutableListOf<IRelayPolicy>()
 
@@ -188,7 +187,11 @@ private fun composePolicy(
     }
 
     if (verifySigs) {
-        pieces += VerifyPolicy
+        // When parallel verify is on, the IngestQueue handles EVENT
+        // verification on the writer's CPU fan-out — but AUTH events
+        // bypass the queue, so we still need the policy chain to
+        // verify those. `VerifyAuthOnlyPolicy` does exactly that.
+        pieces += if (parallelVerify) VerifyAuthOnlyPolicy else VerifyPolicy
     }
 
     return pieces.fold<IRelayPolicy, IRelayPolicy>(EmptyPolicy) { acc, p ->

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/IngestQueue.kt

@@ -30,6 +30,7 @@ import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ClosedReceiveChannelException
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
 import kotlin.coroutines.CoroutineContext
@@ -168,43 +169,53 @@ class IngestQueue(
                 processBatch(batch)
                 batch.clear()
             }
-        } catch (_: kotlinx.coroutines.channels.ClosedReceiveChannelException) {
+        } catch (_: ClosedReceiveChannelException) {
             // Normal shutdown via close().
         }
     }
 
     private suspend fun processBatch(batch: List<Submission>) {
-        // Tier 3: parallel pre-insert verify. Each batch entry that
+        val verifyResults = verifyBatch(batch)
-        // fails verification is pre-marked Rejected and excluded from
+        val finalOutcomes = runInsertStage(batch, verifyResults)
-        // the SQLite transaction. The remaining (verified) events go
+        dispatchOutcomes(batch, finalOutcomes)
-        // through batchInsert in original order; we re-stitch outcomes
+    }
-        // back to the full batch by index.
-        val verifyResults: BooleanArray? =
-            verify?.let { hook ->
-                coroutineScope {
-                    batch
-                        .map { sub -> async(Dispatchers.Default) { hook(sub.event) } }
-                        .awaitAll()
-                        .toBooleanArray()
-                }
-            }
 
-        val toInsert: List<Event>
+    /**
-        val insertIndices: IntArray
+     * Tier 3: per-row verify. Returns `null` when no [verify] hook is
-        if (verifyResults == null) {
+     * configured (skip the stage entirely). For multi-event batches
-            toInsert = batch.map { it.event }
+     * each verify runs as its own `async(Default)` so they spread
-            insertIndices = IntArray(batch.size) { it }
+     * across CPU cores; single-event batches short-circuit to a
-        } else {
+     * direct call to avoid coroutine-scope overhead.
-            val accepted = ArrayList<Event>(batch.size)
+     */
-            val mapping = ArrayList<Int>(batch.size)
+    private suspend fun verifyBatch(batch: List<Submission>): BooleanArray? {
-            for (i in batch.indices) {
+        val hook = verify ?: return null
-                if (verifyResults[i]) {
+        if (batch.size == 1) return BooleanArray(1) { hook(batch[0].event) }
-                    accepted.add(batch[i].event)
+        return coroutineScope {
-                    mapping.add(i)
+            batch
-                }
+                .map { sub -> async(Dispatchers.Default) { hook(sub.event) } }
+                .awaitAll()
+                .toBooleanArray()
+        }
+    }
+
+    /**
+     * Run the SQLite transaction for the verified subset of [batch]
+     * and stitch outcomes back to a per-batch-index array. Failed
+     * verifies pre-mark `Rejected` and skip the insert. A whole-batch
+     * commit failure converts every persisted entry to `Rejected`
+     * with the throw message.
+     */
+    private suspend fun runInsertStage(
+        batch: List<Submission>,
+        verifyResults: BooleanArray?,
+    ): Array<IEventStore.InsertOutcome?> {
+        val toInsert = ArrayList<Event>(batch.size)
+        val insertIndices = ArrayList<Int>(batch.size)
+        for (i in batch.indices) {
+            if (verifyResults == null || verifyResults[i]) {
+                toInsert.add(batch[i].event)
+                insertIndices.add(i)
             }
-            toInsert = accepted
-            insertIndices = mapping.toIntArray()
         }
 
         val insertOutcomes: List<IEventStore.InsertOutcome> =
@@ -220,11 +231,10 @@ class IngestQueue(
                 }
             }
 
-        // Build a per-batch-index outcome array.
         val finalOutcomes = arrayOfNulls<IEventStore.InsertOutcome>(batch.size)
         for (j in insertIndices.indices) {
-            finalOutcomes[insertIndices[j]] = insertOutcomes.getOrNull(j)
+            finalOutcomes[insertIndices[j]] =
-                ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome")
+                insertOutcomes.getOrNull(j) ?: missingOutcome
         }
         if (verifyResults != null) {
             for (i in batch.indices) {
@@ -233,12 +243,16 @@ class IngestQueue(
                 }
             }
         }
+        return finalOutcomes
+    }
 
+    private fun dispatchOutcomes(
+        batch: List<Submission>,
+        outcomes: Array<IEventStore.InsertOutcome?>,
+    ) {
         for (i in batch.indices) {
             val sub = batch[i]
-            val outcome =
+            val outcome = outcomes[i] ?: missingOutcome
-                finalOutcomes[i]
-                    ?: IEventStore.InsertOutcome.Rejected("internal error: missing outcome")
             try {
                 sub.onComplete(outcome)
             } catch (e: Throwable) {
@@ -262,6 +276,14 @@ class IngestQueue(
     }
 
     companion object {
+        /**
+         * Default for a missing per-row outcome — only reachable on a
+         * contract violation (the store returned fewer outcomes than
+         * inserts), so the message is informational, not user-facing.
+         */
+        private val missingOutcome =
+            IEventStore.InsertOutcome.Rejected("internal error: missing outcome")
+
         /**
          * Cap per batch. Sized to keep per-batch latency low (each
          * transaction holds the SQLite writer mutex; over-large

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NostrServer.kt

@@ -20,7 +20,6 @@
  */
 package com.vitorpamplona.quartz.nip01Core.relay.server
 
-import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.crypto.verify
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
@@ -65,7 +64,7 @@ class NostrServer(
         IngestQueue(
             store = store,
             parentContext = parentContext,
-            verify = if (parallelVerify) ::verifyEvent else null,
+            verify = if (parallelVerify) ({ it.verify() }) else null,
         )
 
     private val subStore = LiveEventStore(store, ingest)
@@ -128,11 +127,4 @@ class NostrServer(
      */
     @Deprecated("Use close() instead", replaceWith = ReplaceWith("close()"))
     fun shutdown() = close()
-
-    private companion object {
-        // Function reference (`Event::verify`) wrapper so the
-        // ingest hook keeps a single instance per server rather
-        // than allocating a fresh lambda on every call site.
-        private fun verifyEvent(event: Event): Boolean = event.verify()
-    }
 }

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt

@@ -43,6 +43,7 @@ import com.vitorpamplona.quartz.utils.cache.LargeCache
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.channels.ClosedSendChannelException
 import kotlinx.coroutines.launch
 
 /**
@@ -147,7 +148,7 @@ class RelaySession(
                     }
                 }
             }
-        } catch (_: kotlinx.coroutines.channels.ClosedSendChannelException) {
+        } catch (_: ClosedSendChannelException) {
             // Server is shutting down — the queue is closed. Reply
             // with a transient failure so a client re-trying against
             // the next instance gets a sane signal; the WS itself is

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/VerifyPolicy.kt

@@ -31,13 +31,24 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.IRelayPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.PolicyResult
 
 /**
- * Allows all commands without authentication. This is the default policy.
+ * Verifies the Schnorr signature + id hash of every incoming
+ * `EVENT` and `AUTH` event. Other commands pass through.
+ *
+ * The default `VerifyPolicy` singleton verifies both. The
+ * [VerifyAuthOnlyPolicy] singleton skips the EVENT path — use it
+ * when the [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue]
+ * is doing parallel verify (Tier 3 of the event-ingestion plan)
+ * so EVENTs aren't verified twice. AUTH is still verified inline
+ * because the AUTH path bypasses the queue entirely; without it,
+ * a forged event could mark a pubkey as authenticated.
  */
-object VerifyPolicy : IRelayPolicy {
+open class VerifyEventsAndAuthPolicy(
+    private val verifyEvents: Boolean,
+) : IRelayPolicy {
     override fun onConnect(send: (Message) -> Unit) { }
 
     override fun accept(cmd: EventCmd) =
-        if (cmd.event.verify()) {
+        if (!verifyEvents || cmd.event.verify()) {
             PolicyResult.Accepted(cmd)
         } else {
             PolicyResult.Rejected("invalid: bad signature or id")
@@ -56,3 +67,18 @@ object VerifyPolicy : IRelayPolicy {
 
     override fun canSendToSession(event: Event) = true
 }
+
+/**
+ * Default verify policy — checks every EVENT and every AUTH. Use
+ * this when nothing else in the stack verifies signatures.
+ */
+object VerifyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = true)
+
+/**
+ * Verify policy that skips the EVENT path. Use when the
+ * `IngestQueue` is configured with `parallelVerify`, so EVENTs are
+ * verified once on the writer's CPU fan-out instead of inline on
+ * the WebSocket pump. AUTH is still verified inline because that
+ * command never reaches the queue.
+ */
+object VerifyAuthOnlyPolicy : VerifyEventsAndAuthPolicy(verifyEvents = false)

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt

@@ -137,8 +137,11 @@ class ObservableEventStore(
             }
         }
 
-        @Suppress("UNCHECKED_CAST")
+        // Every index in `events.indices` was populated above (either
-        return outcomes.toList() as List<IEventStore.InsertOutcome>
+        // from the ephemeral pre-pass or the `inner.batchInsert`
+        // result), so no nulls remain. `requireNoNulls()` enforces
+        // that with a runtime check instead of an unchecked cast.
+        return outcomes.requireNoNulls().asList()
     }
 
     override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) {