feat(quartz): LiveNegentropyIndex — always-current (created_at, id) set for NIP-77

Milestone 1 of quartz/plans/2026-07-03-incremental-negentropy-storage.md. Sorted array with binary-search insert (near-tail in the common case), itemized remove for displaced rows, wholesale invalidate for delete paths that can't itemize, and sealed snapshots memoized per mutation generation — reconcile only reads, so one snapshot backs any number of concurrent sessions and stays immutable under later writes. Over-cap answers null so the caller keeps the strfry-parity NEG-ERR. Not wired into any store yet; next milestones plumb displaced-row deltas from the SQLite modules and serve index-total filters from it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TtDNpayEYvJH7QuPswND3A
2026-08-09 08:04:45 +00:00 · 2026-07-03 22:41:35 +00:00
parent 175cac3e46
commit 7ad7dee5fa
2 changed files with 389 additions and 0 deletions
@@ -0,0 +1,215 @@
+/*
+ * 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.nip77Negentropy
+
+import com.vitorpamplona.negentropy.storage.IStorage
+import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
+import kotlin.concurrent.atomics.AtomicBoolean
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
+
+/**
+ * Always-current `(created_at, id)` index for NIP-77 negentropy — the
+ * strfry-parity answer to cold NEG-OPEN cost. Instead of paying a full
+ * table scan + O(n log n) seal on every open (relayBench: ~340 ms at
+ * 50k events; strfry serves ~21 ms off its always-current tree), the
+ * relay maintains this sorted set incrementally from the write path
+ * and a NEG-OPEN only pays one O(n) copy into a sealed snapshot — and
+ * even that is cached until the next mutation.
+ *
+ * Ordering matches negentropy's item order (`StorageUnit.compareTo`):
+ * `created_at` ascending, then id bytes ascending — for lowercase hex
+ * ids, byte order and string order agree, so entries compare by
+ * ([IdAndTime.createdAt], [IdAndTime.id]).
+ *
+ * Lifecycle contract (see
+ * `quartz/plans/2026-07-03-incremental-negentropy-storage.md`):
+ *
+ *  - **Populate** with [rebuild] from one scan, then keep current with
+ *    [insert] / [remove] as the store mutates. Inserts are near-tail in
+ *    the common case (`created_at` ≈ now), so the memmove is tiny.
+ *  - **Removals the caller can't itemize** (kind-5 deletes by filter,
+ *    expiration sweeps, admin purges) call [invalidate] instead; the
+ *    index answers nothing until the next [rebuild]. Correctness rule:
+ *    the index must never advertise an id the store no longer has —
+ *    peers would fetch dead ids — so when in doubt, invalidate.
+ *  - **Snapshot** with [sealedSnapshot]; reconciliation only reads the
+ *    sealed storage, so one snapshot backs any number of concurrent
+ *    sessions and stays valid even if the live index mutates after.
+ *
+ * Thread-safety: all operations take a short spin lock (same pattern as
+ * `LiveEventStore`'s replay dedup). Mutations arrive from the store's
+ * single writer; snapshots from any REQ coroutine.
+ */
+@OptIn(ExperimentalAtomicApi::class)
+class LiveNegentropyIndex {
+    private val lock = AtomicBoolean(false)
+
+    /** Sorted by (createdAt, id). Only touched under [locked]. */
+    private var entries = ArrayList<IdAndTime>()
+
+    /** False until the first [rebuild], and after any [invalidate]. */
+    private var populated = false
+
+    /** Bumped on every mutation; keys the memoized snapshot. */
+    private var generation = 0L
+
+    private var cachedSnapshot: IStorage? = null
+    private var cachedGeneration = -1L
+
+    private inline fun <R> locked(block: () -> R): R {
+        while (lock.exchange(true)) {
+            while (lock.load()) { }
+        }
+        try {
+            return block()
+        } finally {
+            lock.store(false)
+        }
+    }
+
+    private fun compare(
+        a: IdAndTime,
+        b: IdAndTime,
+    ): Int {
+        val byTime = a.createdAt.compareTo(b.createdAt)
+        if (byTime != 0) return byTime
+        return a.id.compareTo(b.id)
+    }
+
+    /**
+     * Binary search for [entry]'s position. Returns the index when
+     * present, or `-(insertionPoint) - 1` when absent — same contract
+     * as `java.util.Collections.binarySearch`.
+     */
+    private fun search(entry: IdAndTime): Int {
+        var low = 0
+        var high = entries.size - 1
+        while (low <= high) {
+            val mid = (low + high) ushr 1
+            val cmp = compare(entries[mid], entry)
+            when {
+                cmp < 0 -> low = mid + 1
+                cmp > 0 -> high = mid - 1
+                else -> return mid
+            }
+        }
+        return -(low + 1)
+    }
+
+    /** True once [rebuild] ran and no [invalidate] happened since. */
+    fun isPopulated(): Boolean = locked { populated }
+
+    fun size(): Int = locked { entries.size }
+
+    /**
+     * Replaces the whole index with [snapshot] (any order; sorted here)
+     * and marks it populated. This is the recovery path after boot or
+     * [invalidate] — one store scan, then incremental maintenance
+     * resumes.
+     */
+    fun rebuild(snapshot: List<IdAndTime>) {
+        val sorted = ArrayList(snapshot)
+        sorted.sortWith(::compare)
+        locked {
+            entries = sorted
+            populated = true
+            generation++
+            cachedSnapshot = null
+        }
+    }
+
+    /**
+     * Drops the index content and answers nothing until the next
+     * [rebuild]. Call from any store mutation whose displaced rows
+     * can't be itemized (delete-by-filter, expiration sweep, vanish).
+     */
+    fun invalidate() {
+        locked {
+            entries = ArrayList()
+            populated = false
+            generation++
+            cachedSnapshot = null
+        }
+    }
+
+    /** Adds one entry. Duplicate `(createdAt, id)` pairs are ignored. */
+    fun insert(entry: IdAndTime) {
+        locked {
+            if (!populated) return
+            val at = search(entry)
+            if (at >= 0) return
+            entries.add(-(at + 1), entry)
+            generation++
+            cachedSnapshot = null
+        }
+    }
+
+    /** Removes one entry; no-op when absent. */
+    fun remove(entry: IdAndTime) {
+        locked {
+            if (!populated) return
+            val at = search(entry)
+            if (at < 0) return
+            entries.removeAt(at)
+            generation++
+            cachedSnapshot = null
+        }
+    }
+
+    /**
+     * A **sealed** negentropy storage of the current content, ready to
+     * back a [NegentropyServerSession]. Memoized per mutation
+     * generation: back-to-back NEG-OPENs with no writes in between (the
+     * mirror-heartbeat pattern) share one snapshot; after a write, only
+     * the first open pays the O(n) copy + seal of already-sorted data.
+     *
+     * Returns `null` when the index is not [isPopulated] (caller falls
+     * back to a scan — and should [rebuild] from it), or when the
+     * content exceeds [maxEntries] (caller answers NEG-ERR, the
+     * strfry-parity `"blocked: too many query results"`).
+     */
+    fun sealedSnapshot(maxEntries: Int): IStorage? {
+        // Fast path reuses the memoized snapshot; the slow path copies
+        // the entries out under the lock and seals OUTSIDE it so a big
+        // seal doesn't stall the writer.
+        var toSeal: ArrayList<IdAndTime>? = null
+        var sealGeneration = 0L
+        locked {
+            if (!populated) return null
+            if (entries.size > maxEntries) return null
+            val cached = cachedSnapshot
+            if (cached != null && cachedGeneration == generation) return cached
+            toSeal = ArrayList(entries)
+            sealGeneration = generation
+        }
+
+        val sealed = NegentropyServerSession.sealVector(toSeal!!)
+        locked {
+            // Last sealer wins; any competing snapshot of the same
+            // generation is equivalent content-wise.
+            if (populated && sealGeneration == generation) {
+                cachedSnapshot = sealed
+                cachedGeneration = sealGeneration
+            }
+        }
+        return sealed
+    }
+}
@@ -0,0 +1,174 @@
+/*
+ * 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.nip77Negentropy
+
+import com.vitorpamplona.negentropy.Negentropy
+import com.vitorpamplona.negentropy.storage.IStorage
+import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
+import com.vitorpamplona.quartz.utils.Hex
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertNotSame
+import kotlin.test.assertNull
+import kotlin.test.assertSame
+import kotlin.test.assertTrue
+
+class LiveNegentropyIndexTest {
+    private fun id(seed: Int): String = seed.toString(16).padStart(64, '0')
+
+    private fun entry(
+        time: Long,
+        seed: Int = time.toInt(),
+    ) = IdAndTime(time, id(seed))
+
+    /** The index content, read back through the sealed storage. */
+    private fun IStorage.toEntries(): List<IdAndTime> = map { IdAndTime(it.timestamp, it.id.toHexString()) }
+
+    private fun contentOf(index: LiveNegentropyIndex): List<IdAndTime> = assertNotNull(index.sealedSnapshot(Int.MAX_VALUE)).toEntries()
+
+    @Test
+    fun answersNothingUntilRebuilt() {
+        val index = LiveNegentropyIndex()
+        assertNull(index.sealedSnapshot(1000))
+
+        // Mutations before the first rebuild are folded into the scan
+        // that populates it — they must not resurrect a dropped index.
+        index.insert(entry(1))
+        assertNull(index.sealedSnapshot(1000))
+        assertEquals(0, index.size())
+    }
+
+    @Test
+    fun rebuildSortsAndSnapshotMatches() {
+        val index = LiveNegentropyIndex()
+        index.rebuild(listOf(entry(5), entry(1), entry(3)))
+
+        assertTrue(index.isPopulated())
+        assertEquals(3, index.size())
+        assertEquals(listOf(entry(1), entry(3), entry(5)), contentOf(index))
+    }
+
+    @Test
+    fun insertsKeepOrderIncludingBackfillAndTies() {
+        val index = LiveNegentropyIndex()
+        index.rebuild(listOf(entry(10), entry(30)))
+
+        index.insert(entry(40)) // tail (the common, cheap case)
+        index.insert(entry(20)) // out-of-order backfill
+        index.insert(IdAndTime(10, id(99))) // same timestamp, distinct id
+
+        assertEquals(
+            listOf(entry(10), IdAndTime(10, id(99)), entry(20), entry(30), entry(40)),
+            contentOf(index),
+        )
+    }
+
+    @Test
+    fun duplicateInsertIsIgnored() {
+        val index = LiveNegentropyIndex()
+        index.rebuild(listOf(entry(1)))
+
+        index.insert(entry(1))
+
+        assertEquals(1, index.size())
+    }
+
+    @Test
+    fun removeDropsExactlyTheEntry() {
+        val index = LiveNegentropyIndex()
+        index.rebuild(listOf(entry(1), entry(2), entry(3)))
+
+        index.remove(entry(2))
+        index.remove(entry(7)) // absent: no-op
+
+        assertEquals(listOf(entry(1), entry(3)), contentOf(index))
+    }
+
+    @Test
+    fun invalidateDropsEverythingUntilNextRebuild() {
+        val index = LiveNegentropyIndex()
+        index.rebuild(listOf(entry(1), entry(2)))
+
+        index.invalidate()
+
+        assertTrue(!index.isPopulated())
+        assertNull(index.sealedSnapshot(1000))
+
+        index.rebuild(listOf(entry(9)))
+        assertEquals(listOf(entry(9)), contentOf(index))
+    }
+
+    @Test
+    fun snapshotIsMemoizedUntilTheNextMutation() {
+        val index = LiveNegentropyIndex()
+        index.rebuild(listOf(entry(1)))
+
+        val first = index.sealedSnapshot(1000)
+        val second = index.sealedSnapshot(1000)
+        assertSame(first, second)
+
+        index.insert(entry(2))
+        val third = index.sealedSnapshot(1000)
+        assertNotSame(first, third)
+    }
+
+    @Test
+    fun snapshotIsImmutableUnderLaterMutations() {
+        val index = LiveNegentropyIndex()
+        index.rebuild(listOf(entry(1), entry(2)))
+
+        val snapshot = assertNotNull(index.sealedSnapshot(1000))
+        index.insert(entry(3))
+        index.remove(entry(1))
+
+        // The reconcile a peer started before those writes still sees
+        // the state it opened against.
+        assertEquals(listOf(entry(1), entry(2)), snapshot.toEntries())
+        assertEquals(listOf(entry(2), entry(3)), contentOf(index))
+    }
+
+    @Test
+    fun overCapAnswersNullSoTheCallerSendsNegErr() {
+        val index = LiveNegentropyIndex()
+        index.rebuild(listOf(entry(1), entry(2), entry(3)))
+
+        assertNull(index.sealedSnapshot(2))
+        assertNotNull(index.sealedSnapshot(3))
+    }
+
+    @Test
+    fun snapshotDrivesARealReconcileSession() {
+        // End-to-end sanity: the sealed snapshot must be a valid server
+        // storage for an actual NegentropyServerSession handshake.
+        val index = LiveNegentropyIndex()
+        index.rebuild((1..50).map { entry(it.toLong()) })
+
+        val storage = assertNotNull(index.sealedSnapshot(1000))
+        val server = NegentropyServerSession("sub", storage)
+
+        // A client with an empty set opens the session; the server's
+        // first response must be a parseable NEG-MSG (non-null).
+        val clientInitial = Negentropy(NegentropyServerSession.sealVector(emptyList())).initiate()
+        val response = server.processMessage(Hex.encode(clientInitial))
+        assertNotNull(response)
+    }
+}