fix(store): rank multi-filter search REQs + harden merge from branch audit

Two-reviewer adversarial audit of the branch found no correctness/data-loss/
crash bugs. This closes one gap and applies three robustness fixes:

- Multi-filter search ordering: a REQ whose filters all carry a search term
  (e.g. the client's search-across-kinds) was created_at-ordered via the union
  path. Now relevance-ordered — unionSubqueriesIfNeeded(projectRank) projects
  rank per branch, UNION ALL + GROUP BY row_id MIN(rank) dedups across branches
  keeping the best score. Only when every branch is a search branch; mixed
  search/non-search REQs and count/delete unions stay as before.
- prepareAuthorStreams/prepareTagStreams build cursors via buildStreams, which
  closes already-prepared statements if a later prepare throws (was: stranded
  checked-out, un-reset handles holding read locks in the pooled connection).
- Stream counts computed as Long so authors×kinds / values×kinds can't overflow
  Int back into the eligible band and route a huge fan-out into the merge.
- Renamed CachedStatement.finalize() -> finalizeStatement(): a no-arg finalize()
  is the JVM Object.finalize, risking a GC double-close of the native handle.

Tests: multi-filter search relevance + cross-branch dedup + count parity;
existing merge/cache/search suites green.

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

quartz/plans/2026-07-21-sqlite-query-scaling.md

@@ -135,6 +135,34 @@ reuse, and cap overflow → uncached fallback.
   whose asserted EXPLAIN output updated for the new join column
   (`event_fts.rowid`) and the contentless table's virtual-index marker
   (`0:M2`→`0:M1`).
-- Search ordering unchanged: still exact `created_at DESC` (NIP-01 `limit`
-  semantics); the delete + size + optimize wins are unconditional and
-  spec-neutral.
+- Search ordering: bm25 relevance for every search REQ shape — tag-free,
+  `search + tag`, and multi-filter all-search (e.g. the client's
+  search-across-kinds, unioned then deduped by event keeping the best score).
+  Mixed search/non-search multi-filter REQs stay `created_at` (a non-search
+  branch has no defined relevance).
+
+## Audit follow-ups (post-review hardening)
+
+A two-reviewer adversarial audit of the branch found no correctness/data-loss/
+crash bugs; it produced these robustness fixes and one gap-closure:
+
+- **Multi-filter search ordering** (gap): a REQ of several filters that all
+  carry a search term was `created_at`-ordered (the union path). Now
+  relevance-ordered via `unionSubqueriesIfNeeded(projectRank)` — each branch
+  projects `rank`, `UNION ALL` + `GROUP BY row_id MIN(rank)` dedups across
+  branches keeping the best score. Only when every branch is a search branch
+  (and FTS is on); count/delete unions stay single-column.
+- **Merge stream-prep leak** (F1): `prepareAuthorStreams`/`prepareTagStreams`
+  now build cursors through `buildStreams`, which closes any already-prepared
+  statements if a later prepare throws — otherwise a mid-loop failure stranded
+  checked-out, un-reset handles (read locks) in the pooled connection.
+- **Stream-count overflow** (F3): `authors×kinds` / `values×kinds` computed as
+  `Long` so a pathological product can't wrap `Int` back into the eligible
+  band and route a huge fan-out into the merge.
+- **Finalizer footgun** (F2): the pooled statement's `finalize()` was renamed
+  `finalizeStatement()` — a no-arg `finalize()` is the JVM's `Object.finalize`,
+  so the GC could double-close the native handle after explicit close.
+
+Migration cost noted (low/operational, not correctness): the synchronous
+v4→v5 rebuild runs in the upgrade transaction; atomic and kill-safe (rolls
+back to v4), but a very large client store pays a one-time first-open stall.

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/FullTextSearchModule.kt

@@ -53,10 +53,12 @@ import com.vitorpamplona.quartz.utils.EventFactory
  * by quality of search result ... not by the usual `.created_at`", limit
  * applied after the score) — via FTS5 bm25 (`ORDER BY event_fts.rank`), with
  * `created_at DESC` only as a tie-break. This holds for *every* search filter:
- * the tag-free shape ([QueryBuilder.makeSimpleSearch]) and `search + tag`
- * (whose row-id subquery carries the rank through via `projectRank`). Only the
- * negentropy snapshot keeps `created_at` — it is a sync set, not a ranked
- * result. bm25 must score every match, so search latency still grows with the
+ * the tag-free shape ([QueryBuilder.makeSimpleSearch]), `search + tag` (whose
+ * row-id subquery carries the rank through via `projectRank`), and a
+ * multi-filter all-search REQ (unioned, deduped by event keeping the best
+ * score). Only the negentropy snapshot — and a multi-filter REQ mixing search
+ * and non-search branches — keep `created_at` (a sync set / a branch with no
+ * defined relevance). bm25 must score every match, so search latency still grows with the
  * match set regardless of ordering; corpus-independent search needs an
  * external engine, not this index.
  *

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt

@@ -125,17 +125,19 @@ internal object MergeQueryExecutor {
         // distinct set.
         val distinctAuthors = authors.distinct().size
         val kinds = filter.kinds
+        // Long product: a pathological authors×kinds could overflow Int and
+        // wrap back into the eligible band, routing a huge fan-out here.
         val streams =
             if (kinds != null && kinds.isNotEmpty()) {
-                distinctAuthors * kinds.distinct().size
+                distinctAuthors.toLong() * kinds.distinct().size
             } else {
                 // authors-only needs the (pubkey, created_at) index to stream.
                 if (!indexStrategy.indexEventsByPubkeyAlone) return -1
-                distinctAuthors
+                distinctAuthors.toLong()
             }
         // A single stream is already the optimal single index seek — let the
         // normal path handle it; only merge when there's something to merge.
-        return if (streams in 2..MAX_STREAMS) streams else -1
+        return if (streams in 2..MAX_STREAMS.toLong()) streams.toInt() else -1
     }
 
     /**
@@ -170,12 +172,12 @@ internal object MergeQueryExecutor {
         val kinds = filter.kinds?.distinct()?.takeIf { it.isNotEmpty() }
         val streams =
             if (kinds != null) {
-                values.size * kinds.size
+                values.size.toLong() * kinds.size
             } else {
                 if (!indexStrategy.indexTagsByCreatedAtAlone) return -1
-                values.size
+                values.size.toLong()
             }
-        return if (streams in 2..MAX_STREAMS) streams else -1
+        return if (streams in 2..MAX_STREAMS.toLong()) streams.toInt() else -1
     }
 
     /** Prepares one bound, newest-first cursor per author stream. */
@@ -199,8 +201,7 @@ internal object MergeQueryExecutor {
         val orderBy =
             if (indexStrategy.useAndIndexIdOnOrderBy) "created_at DESC, id ASC" else "created_at DESC"
 
-        val stmts = ArrayList<SQLiteStatement>((kinds?.size ?: 1) * authors.size)
-        if (kinds != null) {
+        return if (kinds != null) {
             val sql =
                 buildString {
                     append("SELECT ").append(COLS)
@@ -210,16 +211,14 @@ internal object MergeQueryExecutor {
                     if (since != null) append(" AND created_at >= ?")
                     append(" ORDER BY ").append(orderBy)
                 }
-            for (kind in kinds) {
-                for (author in authors) {
+            buildStreams(kinds.size * authors.size) { i ->
                 val stmt = db.prepare(sql)
                 var p = 1
-                    stmt.bindLong(p++, kind.toLong())
-                    stmt.bindText(p++, author)
+                stmt.bindLong(p++, kinds[i / authors.size].toLong())
+                stmt.bindText(p++, authors[i % authors.size])
                 if (until != null) stmt.bindLong(p++, until)
                 if (since != null) stmt.bindLong(p++, since)
-                    stmts.add(stmt)
-                }
+                stmt
             }
         } else {
             val sql =
@@ -231,16 +230,15 @@ internal object MergeQueryExecutor {
                     if (since != null) append(" AND created_at >= ?")
                     append(" ORDER BY ").append(orderBy)
                 }
-            for (author in authors) {
+            buildStreams(authors.size) { i ->
                 val stmt = db.prepare(sql)
                 var p = 1
-                stmt.bindText(p++, author)
+                stmt.bindText(p++, authors[i])
                 if (until != null) stmt.bindLong(p++, until)
                 if (since != null) stmt.bindLong(p++, since)
-                stmts.add(stmt)
+                stmt
             }
         }
-        return stmts
     }
 
     /** Prepares one bound, newest-first cursor per tag-value stream. */
@@ -258,8 +256,7 @@ internal object MergeQueryExecutor {
 
         // The tag cursors stream off event_tags (which has no id column), so
         // the tie order can only be created_at DESC — see the class doc.
-        val stmts = ArrayList<SQLiteStatement>((kinds?.size ?: 1) * values.size)
-        if (kinds != null) {
+        return if (kinds != null) {
             val sql =
                 buildString {
                     append("SELECT ").append(EH_COLS)
@@ -270,17 +267,14 @@ internal object MergeQueryExecutor {
                     if (since != null) append(" AND event_tags.created_at >= ?")
                     append(" ORDER BY event_tags.created_at DESC")
                 }
-            for (value in values) {
-                val tagHash = hasher.hash(tagName, value)
-                for (kind in kinds) {
+            buildStreams(values.size * kinds.size) { i ->
                 val stmt = db.prepare(sql)
                 var p = 1
-                    stmt.bindLong(p++, tagHash)
-                    stmt.bindLong(p++, kind.toLong())
+                stmt.bindLong(p++, hasher.hash(tagName, values[i / kinds.size]))
+                stmt.bindLong(p++, kinds[i % kinds.size].toLong())
                 if (until != null) stmt.bindLong(p++, until)
                 if (since != null) stmt.bindLong(p++, since)
-                    stmts.add(stmt)
-                }
+                stmt
             }
         } else {
             val sql =
@@ -293,17 +287,15 @@ internal object MergeQueryExecutor {
                     if (since != null) append(" AND event_tags.created_at >= ?")
                     append(" ORDER BY event_tags.created_at DESC")
                 }
-            for (value in values) {
-                val tagHash = hasher.hash(tagName, value)
+            buildStreams(values.size) { i ->
                 val stmt = db.prepare(sql)
                 var p = 1
-                stmt.bindLong(p++, tagHash)
+                stmt.bindLong(p++, hasher.hash(tagName, values[i]))
                 if (until != null) stmt.bindLong(p++, until)
                 if (since != null) stmt.bindLong(p++, since)
-                stmts.add(stmt)
+                stmt
             }
         }
-        return stmts
     }
 
     /**
@@ -328,6 +320,32 @@ internal object MergeQueryExecutor {
         }
     }
 
+    /**
+     * Prepares [count] cursors via [prepareOne], closing any already-prepared
+     * statements if a later prepare throws — otherwise a mid-loop failure would
+     * strand checked-out, un-reset handles in the pooled connection (dead
+     * slots holding read locks). On success the caller ([mergeStreams]) owns
+     * closing them.
+     */
+    private inline fun buildStreams(
+        count: Int,
+        prepareOne: (Int) -> SQLiteStatement,
+    ): List<SQLiteStatement> {
+        val stmts = ArrayList<SQLiteStatement>(count)
+        try {
+            for (i in 0 until count) stmts.add(prepareOne(i))
+        } catch (e: Throwable) {
+            for (s in stmts) {
+                try {
+                    s.close()
+                } catch (_: Throwable) {
+                }
+            }
+            throw e
+        }
+        return stmts
+    }
+
     /**
      * Heap-free k-way merge over the prepared [stmts]: repeatedly emits the
      * newest live head (`created_at DESC`, tie `id ASC`) until [limit] rows

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt

@@ -224,7 +224,13 @@ class QueryBuilder(
     ): QuerySpec {
         if (filters.size == 1) return toSql(filters.first(), hasher)
 
-        val rowIdSubqueries = unionSubqueriesIfNeeded(filters, hasher)
+        // A multi-filter search REQ (e.g. the client's search-across-kinds,
+        // all filters sharing one term) must still be NIP-50 relevance-ordered.
+        // Only when EVERY branch is a search branch (and FTS is on, so each has
+        // a rank column) — a non-search branch has no defined relevance, so a
+        // mixed REQ falls back to created_at.
+        val rankSearch = fts.enabled && filters.all { !it.search.isNullOrEmpty() }
+        val rowIdSubqueries = unionSubqueriesIfNeeded(filters, hasher, projectRank = rankSearch)
 
         return if (rowIdSubqueries == null) {
             QuerySpec(
@@ -233,7 +239,7 @@ class QueryBuilder(
             )
         } else {
             QuerySpec(
-                makeQueryIn(rowIdSubqueries.sql),
+                makeQueryIn(rowIdSubqueries.sql, orderByRank = rankSearch),
                 rowIdSubqueries.args,
             )
         }
@@ -723,16 +729,34 @@ class QueryBuilder(
     fun unionSubqueriesIfNeeded(
         filters: List<Filter>,
         hasher: TagNameValueHasher,
+        // See [prepareRowIDSubQueries]. When set, every branch is a search
+        // branch that also projects a `rank` column; the union keeps one row
+        // per event with its BEST (min) bm25 score so the caller can present
+        // the whole multi-filter search REQ NIP-50-ranked. Callers must only
+        // pass true when all filters carry a search term (else a branch has no
+        // rank column). Off for count/delete, which stay single-column.
+        projectRank: Boolean = false,
     ): QuerySpec? {
         val inner =
             filters.mapNotNull { filter ->
-                prepareRowIDSubQueries(filter, hasher)
+                prepareRowIDSubQueries(filter, hasher, projectRank)
             }
 
         if (inner.isEmpty()) return null
 
-        return if (inner.size == 1) {
-            inner.first()
+        if (inner.size == 1) return inner.first()
+
+        return if (projectRank) {
+            // UNION ALL keeps every (row_id, rank) so an event matching two
+            // branches under different terms isn't dropped before MIN; GROUP BY
+            // then dedups by event keeping the best score.
+            QuerySpec(
+                sql =
+                    "SELECT row_id, MIN(rank) as rank FROM (\n            " +
+                        inner.joinToString("\n            UNION ALL\n            ") { "SELECT row_id, rank FROM (${it.sql})" } +
+                        "\n            ) GROUP BY row_id",
+                args = inner.flatMap { it.args },
+            )
         } else {
             QuerySpec(
                 sql = inner.joinToString("\n            UNION\n            ") { "SELECT row_id FROM (${it.sql})" },

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StatementCachingConnection.kt

@@ -97,7 +97,7 @@ class StatementCachingConnection(
     }
 
     override fun close() {
-        cache.values.forEach { pool -> pool.forEach { runCatching { it.finalize() } } }
+        cache.values.forEach { pool -> pool.forEach { runCatching { it.finalizeStatement() } } }
         cache.clear()
         cachedCount = 0
         delegate.close()
@@ -120,6 +120,10 @@ class StatementCachingConnection(
             checkedOut = false
         }
 
-        fun finalize() = delegate.close()
+        // Not named `finalize`: a no-arg `finalize()` is treated by the JVM as
+        // Object.finalize(), so the GC would call it and double-close the
+        // native handle after our explicit close(). This is only ever invoked
+        // explicitly from the connection's close().
+        fun finalizeStatement() = delegate.close()
     }
 }

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SearchRelevanceOrderTest.kt

@@ -115,6 +115,63 @@ class SearchRelevanceOrderTest {
             "0".repeat(128),
         )
 
+    private fun evk(
+        content: String,
+        createdAt: Long,
+        kind: Int,
+    ): Event = EventFactory.create(hexId(++idSeq), "00".repeat(32), createdAt, kind, arrayOf(), content, "0".repeat(128))
+
+    @Test
+    fun multiFilterSearchIsRelevanceOrderedAcrossBranches() =
+        runBlocking {
+            val store = EventStore(dbName = null)
+            try {
+                // Relevance A > B > C, created_at A < B < C (reverse), and the
+                // branches split by kind: A,C are kind 1 (TextNote), B is kind
+                // 1111 (Comment) — both searchable kinds.
+                val a = evk("apple apple apple apple", 1, 1)
+                val b = evk("apple apple apple", 2, 1111)
+                val c = evk("apple filler filler filler filler filler", 3, 1)
+                store.batchInsert(listOf(a, b, c))
+
+                // The client's search-across-kinds shape: one term, two filters.
+                val filters =
+                    listOf(
+                        Filter(search = "apple", kinds = listOf(1), limit = 100),
+                        Filter(search = "apple", kinds = listOf(1111), limit = 100),
+                    )
+                val ids = store.query<Event>(filters).map { it.id }
+                assertEquals(listOf(a.id, b.id, c.id), ids, "multi-filter search must be relevance-ordered across branches")
+            } finally {
+                store.close()
+            }
+        }
+
+    @Test
+    fun multiFilterSearchDedupsEventsMatchingSeveralBranches() =
+        runBlocking {
+            val store = EventStore(dbName = null)
+            try {
+                val x = evk("banana banana banana", 1, 1)
+                val y = evk("banana one", 2, 1)
+                store.batchInsert(listOf(x, y))
+
+                // Overlapping branches: both kind-1 events match BOTH filters.
+                // GROUP BY row_id must fold each to a single ranked row.
+                val filters =
+                    listOf(
+                        Filter(search = "banana", kinds = listOf(1, 6), limit = 100),
+                        Filter(search = "banana", kinds = listOf(1, 2), limit = 100),
+                    )
+                val ids = store.query<Event>(filters).map { it.id }
+                assertEquals(ids.size, ids.toSet().size, "no event may appear twice across branches")
+                assertEquals(listOf(x.id, y.id), ids, "deduped, relevance-ordered")
+                assertEquals(2, store.count(filters), "count parity across the union")
+            } finally {
+                store.close()
+            }
+        }
+
     @Test
     fun searchWithATagIsAlsoRelevanceOrdered() =
         runBlocking {

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/AdversarialSearchRankTest.kt

@@ -1,131 +0,0 @@
-/*
- * Copyright (c) 2025 Vitor Pamplona
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy of
- * this software and associated documentation files (the "Software"), to deal in
- * the Software without restriction, including without limitation the rights to use,
- * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
- * Software, and to permit persons to whom the Software is furnished to do so,
- * subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in all
- * copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
- * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
- * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
- * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
- * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
- */
-package com.vitorpamplona.quartz.nip01Core.store.sqlite
-
-import com.vitorpamplona.quartz.nip01Core.core.Event
-import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
-import com.vitorpamplona.quartz.utils.EventFactory
-import kotlinx.coroutines.runBlocking
-import kotlin.test.Test
-import kotlin.test.assertEquals
-import kotlin.test.assertTrue
-
-class AdversarialSearchRankTest {
-    private var idSeq = 0
-
-    private fun hexId(n: Int): String {
-        val s = n.toString(16)
-        return "0".repeat(64 - s.length) + s
-    }
-
-    private fun ev(
-        content: String,
-        createdAt: Long,
-        tags: Array<Array<String>>,
-        kind: Int = 1,
-    ): Event =
-        EventFactory.create(
-            hexId(++idSeq),
-            "00".repeat(32),
-            createdAt,
-            kind,
-            tags,
-            content,
-            "0".repeat(128),
-        )
-
-    // A filter with TWO tag keys + search → self-join on event_tags + rank projection.
-    @Test
-    fun twoTagKeysPlusSearchRanksAndDoesNotDuplicate() =
-        runBlocking {
-            val store = EventStore(dbName = null)
-            try {
-                // has both t=nostr and e=<x>, strong match
-                val strong = ev("nostr nostr nostr", 1, arrayOf(arrayOf("t", "nostr"), arrayOf("e", "aa".repeat(32))))
-                // has both tags, weak match (newer)
-                val weak = ev("nostr among many other filler words here padding", 2, arrayOf(arrayOf("t", "nostr"), arrayOf("e", "aa".repeat(32))))
-                // matches term + t but MISSING e tag → excluded by tagsAll-like AND of two keys
-                val missingE = ev("nostr nostr nostr nostr", 3, arrayOf(arrayOf("t", "nostr")))
-                store.batchInsert(listOf(strong, weak, missingE))
-
-                val filter =
-                    Filter(
-                        search = "nostr",
-                        tags = mapOf("t" to listOf("nostr"), "e" to listOf("aa".repeat(32))),
-                        limit = 10,
-                    )
-                val ids = store.query<Event>(filter).map { it.id }
-                // relevance order, both tags required, no duplicate rows
-                assertEquals(listOf(strong.id, weak.id), ids, "two-tag + search must rank and require both tags with no dupes")
-
-                // count parity with query for the same filter
-                assertEquals(ids.size, store.count(filter), "count must equal query size for two-tag search")
-            } finally {
-                store.close()
-            }
-        }
-
-    // count(filter) vs query(filter).size for a search+tag filter WITH a limit
-    // smaller than the match set — NIP-45 parity.
-    @Test
-    fun countEqualsQuerySizeForSearchTagWithLimit() =
-        runBlocking {
-            val store = EventStore(dbName = null)
-            try {
-                val a = ev("apple apple apple", 1, arrayOf(arrayOf("t", "fruit")))
-                val b = ev("apple apple filler", 2, arrayOf(arrayOf("t", "fruit")))
-                val c = ev("apple filler filler words here", 3, arrayOf(arrayOf("t", "fruit")))
-                store.batchInsert(listOf(a, b, c))
-
-                val filter = Filter(search = "apple", tags = mapOf("t" to listOf("fruit")), limit = 2)
-                val qSize = store.query<Event>(filter).size
-                val cnt = store.count(filter)
-                assertEquals(2, qSize, "query must honor limit")
-                assertEquals(qSize, cnt, "count must match query size under a limit (NIP-45)")
-            } finally {
-                store.close()
-            }
-        }
-
-    // Deleting a NON-searchable event (never had an FTS row) fires the
-    // contentless delete trigger against a non-existent rowid.
-    @Test
-    fun deletingNonSearchableEventDoesNotCrash() =
-        runBlocking {
-            val store = EventStore(dbName = null)
-            try {
-                // kind 7 reaction is not a SearchableEvent → no FTS row inserted.
-                val reaction = ev("+", 1, arrayOf(arrayOf("e", "bb".repeat(32))), kind = 7)
-                val note = ev("hello searchable world", 2, arrayOf())
-                store.batchInsert(listOf(reaction, note))
-
-                // Trigger fires DELETE FROM event_fts WHERE rowid = <reaction row_id>
-                // which never existed. Must be a harmless no-op.
-                store.store.delete(reaction.id)
-
-                // Searchable note still findable; store intact.
-                assertEquals(note.id, store.query<Event>(Filter(search = "searchable")).single().id)
-                assertTrue(store.query<Event>(Filter(ids = listOf(reaction.id))).isEmpty())
-            } finally {
-                store.close()
-            }
-        }
-}