laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-12 01:07:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(quartz): NIP-45 approximate COUNT — HyperLogLog construction + wire support

Browse Source 
The HLL aggregation side (estimate/merge/encode) existed but the construction
side did not, and the Jackson wire (de)serializer silently dropped the `hll`
field — so a JVM/Android relay could not actually answer COUNT with HLL.

- HyperLogLog.addPubKey(): the NIP-45 construction (register index = pubkey
  byte at the filter offset; value = leading-zero-bits from offset+1, +1),
  KMP-safe. Plus HyperLogLog.builderFor(filter) and an HllBuilder that streams
  event pubkeys into registers and yields an approximate CountResult.
- Plumb CountResult through the count path: SessionBackend.countResult /
  ReqResponder.countResult (default = exact count(); override for approximate/
  hll); RelaySession sends the returned CountResult.
- Fix CountResultSerializer/Deserializer (jvmAndroid) to write/read the `hll`
  hex field, matching the kotlinx serializer the native targets already use.
- Tests for construction (index/value/merge-idempotence) and the wire path;
  RELAY.md Approximate COUNT section.

https://claude.ai/code/session_016YBS2pWCBSDgAMthHzfCTr
This commit is contained in:
Claude
2026-06-03 18:29:01 +00:00
parent 319ffb729a
commit 952da685d8
11 changed files with 309 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/RELAY.md
+22
View File
@@ -312,6 +312,28 @@ class MyRelayTest {
}
```
## Approximate COUNT (NIP-45 HyperLogLog)
`COUNT` is answered by `SessionBackend.countResult(filters)` (and
`ReqResponder.countResult`), which defaults to an exact count. To return a
mergeable HyperLogLog estimate instead — for the six canonical NIP-45 queries
(reaction/repost/quote/reply/comment/follower counts) — fold matching pubkeys
into an `HllBuilder` and return its `CountResult`:
```kotlin
override suspend fun countResult(filters: List<Filter>): CountResult {
val filter = filters.first()
val hll = HyperLogLog.builderFor(filter) ?: return CountResult(count(filters))
store.query(filter) { event -> hll.add(event.pubKey) }
return hll.toCountResult() // count = estimate, approximate = true, hll = registers
}
```
The engine frames `count`/`approximate`/`hll` onto the wire
(`["COUNT", id, {"count":N,"hll":"<512-hex>"}]`). Register arrays built by two
relays over the same corpus merge with `HyperLogLog.merge(...)` into a
deduplicated cross-relay estimate.
## Observability
Both servers take an optional `RelayConnectionListener` and expose a live
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt
+3 -4
View File
@@ -23,7 +23,6 @@ package com.vitorpamplona.quartz.nip01Core.relay.server
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.MachineReadablePrefix
@@ -180,9 +179,9 @@ class RelaySession(
// Policy may rewrite filters to match the user's access level.
val filters = (result as PolicyResult.Accepted).cmd.filters
val total =
val countResult =
try {
store.count(filters)
store.countResult(filters)
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
@@ -190,7 +189,7 @@ class RelaySession(
return
}
send(CountMessage(cmd.queryId, CountResult(total)))
send(CountMessage(cmd.queryId, countResult))
}
// -- NIP-42: AUTH ---------------------------------------------------------
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ReqResponder.kt
+9
View File
@@ -21,6 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.server
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.count
@@ -73,4 +74,12 @@ interface ReqResponder {
* every event.
*/
suspend fun count(filters: List<Filter>): Int = respond(filters).count()
/**
* Answers a NIP-45 COUNT, optionally approximate and/or carrying a
* HyperLogLog payload. The default wraps [count] as an exact result;
* override to return `approximate`/`hll` (see
* [com.vitorpamplona.quartz.nip45Count.HllBuilder]).
*/
suspend fun countResult(filters: List<Filter>): CountResult = CountResult(count(filters))
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ReqResponderBackend.kt
+3
View File
@@ -21,6 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.server
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import kotlinx.coroutines.flow.collect
@@ -44,4 +45,6 @@ class ReqResponderBackend(
}
override suspend fun count(filters: List<Filter>): Int = responder.count(filters)
override suspend fun countResult(filters: List<Filter>): CountResult = responder.countResult(filters)
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/SessionBackend.kt
+10 -1
View File
@@ -21,6 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.server
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
@@ -55,9 +56,17 @@ interface SessionBackend {
onEose: () -> Unit,
)
/** Answers a NIP-45 COUNT. */
/** Answers a NIP-45 COUNT with an exact cardinality. */
suspend fun count(filters: List<Filter>): Int
/**
* Answers a NIP-45 COUNT, allowing an approximate result and/or a
* HyperLogLog register payload (see [com.vitorpamplona.quartz.nip45Count.HllBuilder]).
* The default returns the exact [count] with `approximate = false`; override
* to return `approximate`/`hll`.
*/
suspend fun countResult(filters: List<Filter>): CountResult = CountResult(count(filters))
/**
* Handles an EVENT publish, reporting the per-event outcome through
* [onComplete]. The default rejects — a relay with no store does not
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip45Count/HllBuilder.kt
+77
View File
@@ -0,0 +1,77 @@
/*
* 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.nip45Count
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult
import com.vitorpamplona.quartz.utils.Hex
/**
* Accumulates a NIP-45 HyperLogLog register array from the pubkeys of the
* events that match a COUNT filter, so a relay can answer with a mergeable
* `hll` field instead of (or alongside) an exact count.
*
* Build one with [HyperLogLog.builderFor] (which derives the [offset] from the
* filter), fold every matching event's pubkey in with [add], then read
* [toCountResult] / [build]:
*
* ```
* val hll = HyperLogLog.builderFor(filter) ?: return CountResult(exactCount)
* store.query(filter) { event -> hll.add(event.pubKey) }
* return hll.toCountResult()
* ```
*
* Two relays counting the same corpus with the same filter produce register
* arrays that merge (via [HyperLogLog.merge]) into a deduplicated estimate —
* the whole point of NIP-45 HLL.
*/
class HllBuilder(
val offset: Int,
) {
private val registers = ByteArray(HyperLogLog.NUM_REGISTERS)
/** Folds a 32-byte event pubkey into the registers. */
fun add(pubKey: ByteArray): HllBuilder {
HyperLogLog.addPubKey(registers, pubKey, offset)
return this
}
/** Folds a 64-char hex event pubkey into the registers; no-op if malformed. */
fun add(pubKeyHex: String): HllBuilder {
val bytes =
try {
Hex.decode(pubKeyHex)
} catch (_: Exception) {
return this
}
return add(bytes)
}
/** A copy of the current 256-byte register array. */
fun build(): ByteArray = registers.copyOf()
/** The HyperLogLog cardinality estimate of the folded pubkeys. */
fun estimate(): Int = HyperLogLog.estimate(registers).toInt()
/** An approximate [CountResult] carrying both the estimate and the registers. */
fun toCountResult(): CountResult = CountResult(estimate(), approximate = true, hll = build())
companion object
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip45Count/HyperLogLog.kt
+52
View File
@@ -115,6 +115,52 @@ object HyperLogLog {
*/
fun encode(registers: ByteArray): String = Hex.encode(registers)
/**
* Folds one event's pubkey into [registers] per the NIP-45 construction
* algorithm: the byte at [offset] of the pubkey is the register index, and
* the value is the number of leading zero bits starting at byte `offset+1`
* plus one (kept only if larger than the register's current value).
*
* [registers] must be [NUM_REGISTERS] bytes; [pubKey] is the 32-byte event
* pubkey; [offset] comes from [computeOffset] (range 8..23). Pubkeys shorter
* than `offset+1` bytes are ignored.
*/
fun addPubKey(
registers: ByteArray,
pubKey: ByteArray,
offset: Int,
) {
if (offset < 0 || offset >= pubKey.size) return
val ri = pubKey[offset].toInt() and 0xFF
val value = (leadingZeroBits(pubKey, offset + 1) + 1).coerceAtMost(0xFF)
if (value > (registers[ri].toInt() and 0xFF)) registers[ri] = value.toByte()
}
/**
* Counts consecutive zero bits (MSB first) from byte [fromByteIndex] to the
* end of [bytes]. KMP-safe (no `Integer.numberOfLeadingZeros`).
*/
private fun leadingZeroBits(
bytes: ByteArray,
fromByteIndex: Int,
): Int {
var count = 0
for (i in fromByteIndex until bytes.size) {
val b = bytes[i].toInt() and 0xFF
if (b == 0) {
count += 8
} else {
var mask = 0x80
while (mask != 0 && (b and mask) == 0) {
count++
mask = mask shr 1
}
return count
}
}
return count
}
/**
* Computes the deterministic offset for a given filter, as specified
* by NIP-45. The offset determines which byte of event pubkeys is
@@ -140,6 +186,12 @@ object HyperLogLog {
return hexValue + 8
}
/**
* Creates an [HllBuilder] for [filter], or null when the filter carries no
* tag attribute to derive an [computeOffset] from (i.e. HLL doesn't apply).
*/
fun builderFor(filter: Filter): HllBuilder? = computeOffset(filter)?.let { HllBuilder(it) }
/**
* Extracts the first value from the first tag attribute in the filter.
*/
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/ReqResponderServerTest.kt
+30
View File
@@ -23,13 +23,16 @@ package com.vitorpamplona.quartz.nip01Core.relay.server
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.AuthMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EventMessage
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy
import com.vitorpamplona.quartz.nip45Count.HllBuilder
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.test.UnconfinedTestDispatcher
@@ -107,6 +110,33 @@ class ReqResponderServerTest {
}
}
@Test
fun approximateCountWithHllReachesTheWire() =
runTest {
val dispatcher = UnconfinedTestDispatcher(testScheduler)
val responder =
object : ReqResponder {
override fun respond(filters: List<Filter>): Flow<Event> = flowOf(event(1), event(2))
override suspend fun countResult(filters: List<Filter>): CountResult {
val hll = HllBuilder(offset = 8)
respond(filters).collect { hll.add(it.pubKey) }
return hll.toCountResult()
}
}
ReqResponderServer(responder, parentContext = dispatcher).use { server ->
val collector = MessageCollector()
val session = server.connect(collector.send)
session.receive("""["COUNT","q1",{"kinds":[1]}]""")
val counts = collector.containing("COUNT")
assertEquals(1, counts.size)
assertTrue(counts[0].contains("\"hll\""))
assertTrue(counts[0].contains("\"approximate\":true"))
}
}
@Test
fun eventPublishIsRejected() =
runTest {
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip45Count/HllBuilderTest.kt
+91
View File
@@ -0,0 +1,91 @@
/*
* 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.nip45Count
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
class HllBuilderTest {
/** Builds a 32-byte array (64 hex chars) with the given byte overrides. */
private fun pubkey(vararg overrides: Pair<Int, Int>): ByteArray {
val b = ByteArray(32)
for ((i, v) in overrides) b[i] = v.toByte()
return b
}
@Test
fun registerIndexIsByteAtOffset() {
val regs = ByteArray(HyperLogLog.NUM_REGISTERS)
// offset 8: index = byte[8]=0x05 ; value = leadingZeros(from byte 9) + 1.
// byte[9]=0x80 (1000_0000) -> 0 leading zero bits -> value 1.
HyperLogLog.addPubKey(regs, pubkey(8 to 0x05, 9 to 0x80), offset = 8)
assertEquals(1, regs[5].toInt() and 0xFF)
}
@Test
fun valueCountsLeadingZeroBitsAcrossBytes() {
val regs = ByteArray(HyperLogLog.NUM_REGISTERS)
// index = byte[8]=0x00 -> register 0.
// byte[9]=0x00 (8 zero bits) then byte[10]=0x40 (0100_0000 -> 1 leading zero)
// => 8 + 1 = 9 leading zero bits, +1 => value 10.
HyperLogLog.addPubKey(regs, pubkey(8 to 0x00, 10 to 0x40), offset = 8)
assertEquals(10, regs[0].toInt() and 0xFF)
}
@Test
fun keepsTheLargerValuePerRegister() {
val regs = ByteArray(HyperLogLog.NUM_REGISTERS)
// Both map to register 0x00; first yields a larger value than the second.
HyperLogLog.addPubKey(regs, pubkey(8 to 0x00, 10 to 0x40), offset = 8) // value 10
HyperLogLog.addPubKey(regs, pubkey(8 to 0x00, 9 to 0x80), offset = 8) // value 1
assertEquals(10, regs[0].toInt() and 0xFF)
}
@Test
fun builderProducesApproximateCountResultWithRegisters() {
val builder = HllBuilder(offset = 8)
builder.add(pubkey(8 to 0x01, 9 to 0x80))
builder.add(pubkey(8 to 0x02, 9 to 0x80))
val result = builder.toCountResult()
assertTrue(result.approximate)
assertEquals(HyperLogLog.NUM_REGISTERS, result.hll?.size)
assertEquals(1, result.hll!![1].toInt() and 0xFF)
assertEquals(1, result.hll!![2].toInt() and 0xFF)
}
@Test
fun twoBuildersOverSameCorpusMergeToSameRegisters() {
val a = HllBuilder(8).add(pubkey(8 to 0x01, 9 to 0x80)).build()
val b = HllBuilder(8).add(pubkey(8 to 0x01, 9 to 0x80)).build()
val merged = HyperLogLog.merge(listOf(a, b))
// Idempotent: the same pubkey counted twice estimates as one element.
assertTrue(merged.contentEquals(a))
}
@Test
fun malformedHexIsIgnored() {
val builder = HllBuilder(8)
builder.add("not-hex")
assertTrue(builder.build().all { it.toInt() == 0 })
}
}
quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CountResultDeserializer.kt
+9
View File
@@ -21,6 +21,7 @@
package com.vitorpamplona.quartz.nip01Core.relay.commands.toClient
import com.fasterxml.jackson.databind.JsonNode
import com.vitorpamplona.quartz.utils.Hex
class CountResultDeserializer {
companion object {
@@ -28,6 +29,14 @@ class CountResultDeserializer {
CountResult(
count = jsonObject.get("count")?.asInt() ?: 0,
approximate = jsonObject.get("approximate")?.asBoolean() ?: false,
hll =
jsonObject.get("hll")?.asText()?.let {
try {
Hex.decode(it)
} catch (_: Exception) {
null
}
},
)
}
}
quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toClient/CountResultSerializer.kt
+3 -1
View File
@@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.commands.toClient
import com.fasterxml.jackson.core.JsonGenerator
import com.fasterxml.jackson.databind.SerializerProvider
import com.fasterxml.jackson.databind.ser.std.StdSerializer
import com.vitorpamplona.quartz.utils.Hex
class CountResultSerializer : StdSerializer<CountResult>(CountResult::class.java) {
override fun serialize(
@@ -30,12 +31,13 @@ class CountResultSerializer : StdSerializer<CountResult>(CountResult::class.java
gen: JsonGenerator,
provider: SerializerProvider,
) {
// NIP-45 result object: { "count": <int>, "approximate": <bool>? }.
// NIP-45 result object: { "count": <int>, "approximate": <bool>?, "hll": <512-hex>? }.
gen.writeStartObject()
gen.writeNumberField("count", result.count)
if (result.approximate) {
gen.writeBooleanField("approximate", true)
}
result.hll?.let { gen.writeStringField("hll", Hex.encode(it)) }
gen.writeEndObject()
}
}