feat(negentropy): strfry-parity NIP-77 reconciliation path
Implements the A+B+C+D plan in geode/plans/2026-05-07-negentropy-large-corpus.md:
- A: id-and-time-only snapshot. Adds IEventStore.snapshotIdsForNegentropy
returning IdAndTime(createdAt, id) with no Event materialization. SQLite
override projects directly off event_headers (~40 B/entry instead of
~1 KB/entry; matches strfry's MemoryView footprint).
- B: snapshot-size cap. New [negentropy] config section with
max_sync_events=1_000_000 (mirrors strfry's relay__negentropy__maxSyncEvents).
Overflow returns NEG-ERR "blocked: too many query results" (strfry-exact
wording). No default since-window (strfry honors filters as-is; bounding
silently would break interop).
- C: 500_000-byte frame cap. NegentropyServerSession.DEFAULT_FRAME_SIZE_LIMIT
matches strfry's hard-coded Negentropy ne(storage, 500'000). Configurable
via [negentropy].frame_size_limit.
- D: per-connection session cap = 200 (matches strfry). Overflow sends NOTICE
"too many concurrent NEG requests" (strfry parity).
Error-string interop:
- NEG-MSG with unknown subId → "closed: unknown subscription handle"
- reconcile() parse failure → "PROTOCOL-ERROR"
- snapshot overflow → "blocked: too many query results"
- per-conn cap → NOTICE "too many concurrent NEG requests"
NegentropySettings flows: RelayConfig.NegentropySection → Relay → NostrServer
→ RelaySession → NegSessionRegistry. RelayHub takes optional settings for
tests that need to exercise the caps.
Tests:
- SnapshotIdsForNegentropyTest covers projection correctness across all
indexing strategies + the maxEntries+1 sentinel contract.
- Nip77NegentropyTest gains negOpenSnapshotOverflowReturnsStrFryNegErr
and negOpenPerConnectionCapEmitsNotice.
- Existing negMsgWithoutOpenReturnsNegErr updated to assert strfry wording.
This commit is contained in:
@@ -32,6 +32,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyAuthOnlyPo
|
|||||||
import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy
|
import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
|
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
|
||||||
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||||
import java.io.File
|
import java.io.File
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -109,6 +110,12 @@ fun main(args: Array<String>) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
val stateFile = config.admin.state_file?.let { File(it) }
|
val stateFile = config.admin.state_file?.let { File(it) }
|
||||||
|
val negentropySettings =
|
||||||
|
NegentropySettings(
|
||||||
|
frameSizeLimit = config.negentropy.frame_size_limit,
|
||||||
|
maxSyncEvents = config.negentropy.max_sync_events,
|
||||||
|
maxSessionsPerConnection = config.negentropy.max_sessions_per_connection,
|
||||||
|
)
|
||||||
val relay =
|
val relay =
|
||||||
Relay(
|
Relay(
|
||||||
advertisedUrl,
|
advertisedUrl,
|
||||||
@@ -117,6 +124,7 @@ fun main(args: Array<String>) {
|
|||||||
policyBuilder,
|
policyBuilder,
|
||||||
stateFile = stateFile,
|
stateFile = stateFile,
|
||||||
parallelVerify = parallelVerify,
|
parallelVerify = parallelVerify,
|
||||||
|
negentropySettings = negentropySettings,
|
||||||
)
|
)
|
||||||
// Frame cap honors max_ws_frame_bytes when set; max_ws_message_bytes
|
// Frame cap honors max_ws_frame_bytes when set; max_ws_message_bytes
|
||||||
// is treated as the same cap (Ktor's WebSockets plugin only exposes
|
// is treated as the same cap (Ktor's WebSockets plugin only exposes
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
|
|||||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
|
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
|
||||||
import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation
|
import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation
|
||||||
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||||
import com.vitorpamplona.quartz.nip86RelayManagement.server.BanListPolicy
|
import com.vitorpamplona.quartz.nip86RelayManagement.server.BanListPolicy
|
||||||
import com.vitorpamplona.quartz.nip86RelayManagement.server.BanStore
|
import com.vitorpamplona.quartz.nip86RelayManagement.server.BanStore
|
||||||
import kotlinx.coroutines.SupervisorJob
|
import kotlinx.coroutines.SupervisorJob
|
||||||
@@ -84,6 +85,11 @@ class Relay(
|
|||||||
* `Main.kt` skips `VerifyPolicy` when this flag is on.
|
* `Main.kt` skips `VerifyPolicy` when this flag is on.
|
||||||
*/
|
*/
|
||||||
parallelVerify: Boolean = false,
|
parallelVerify: Boolean = false,
|
||||||
|
/**
|
||||||
|
* NIP-77 server-side tuning (frame cap, snapshot cap,
|
||||||
|
* per-connection session cap). Defaults to strfry-parity values.
|
||||||
|
*/
|
||||||
|
negentropySettings: NegentropySettings = NegentropySettings.Default,
|
||||||
) : AutoCloseable {
|
) : AutoCloseable {
|
||||||
private val stateStore: RelayStateStore? = stateFile?.let { RelayStateStore(it) }
|
private val stateStore: RelayStateStore? = stateFile?.let { RelayStateStore(it) }
|
||||||
|
|
||||||
@@ -168,8 +174,9 @@ class Relay(
|
|||||||
val user = policyBuilder()
|
val user = policyBuilder()
|
||||||
if (user === EmptyPolicy) BanListPolicy(banStore) else user + BanListPolicy(banStore)
|
if (user === EmptyPolicy) BanListPolicy(banStore) else user + BanListPolicy(banStore)
|
||||||
},
|
},
|
||||||
parentContext,
|
parentContext = parentContext,
|
||||||
parallelVerify = parallelVerify,
|
parallelVerify = parallelVerify,
|
||||||
|
negentropySettings = negentropySettings,
|
||||||
)
|
)
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
|
|||||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket
|
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
|
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
|
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
|
||||||
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||||
import java.util.concurrent.ConcurrentHashMap
|
import java.util.concurrent.ConcurrentHashMap
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -50,6 +51,7 @@ import java.util.concurrent.ConcurrentHashMap
|
|||||||
*/
|
*/
|
||||||
class RelayHub(
|
class RelayHub(
|
||||||
private val defaultPolicy: () -> IRelayPolicy = { EmptyPolicy },
|
private val defaultPolicy: () -> IRelayPolicy = { EmptyPolicy },
|
||||||
|
private val negentropySettings: NegentropySettings = NegentropySettings.Default,
|
||||||
) : WebsocketBuilder,
|
) : WebsocketBuilder,
|
||||||
AutoCloseable {
|
AutoCloseable {
|
||||||
private val relays = ConcurrentHashMap<NormalizedRelayUrl, Relay>()
|
private val relays = ConcurrentHashMap<NormalizedRelayUrl, Relay>()
|
||||||
@@ -60,7 +62,11 @@ class RelayHub(
|
|||||||
fun getOrCreate(url: NormalizedRelayUrl): Relay {
|
fun getOrCreate(url: NormalizedRelayUrl): Relay {
|
||||||
check(!closed) { "RelayHub has been closed" }
|
check(!closed) { "RelayHub has been closed" }
|
||||||
return relays.getOrPut(url) {
|
return relays.getOrPut(url) {
|
||||||
Relay(url = url, policyBuilder = defaultPolicy)
|
Relay(
|
||||||
|
url = url,
|
||||||
|
policyBuilder = defaultPolicy,
|
||||||
|
negentropySettings = negentropySettings,
|
||||||
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -43,6 +43,7 @@ data class RelayConfig(
|
|||||||
val limits: LimitsSection = LimitsSection(),
|
val limits: LimitsSection = LimitsSection(),
|
||||||
val authorization: AuthorizationSection = AuthorizationSection(),
|
val authorization: AuthorizationSection = AuthorizationSection(),
|
||||||
val admin: AdminSection = AdminSection(),
|
val admin: AdminSection = AdminSection(),
|
||||||
|
val negentropy: NegentropySection = NegentropySection(),
|
||||||
) {
|
) {
|
||||||
/**
|
/**
|
||||||
* Maps the `[info]` section into a [RelayInfo] used by the NIP-11
|
* Maps the `[info]` section into a [RelayInfo] used by the NIP-11
|
||||||
@@ -160,6 +161,34 @@ data class RelayConfig(
|
|||||||
val max_ws_frame_bytes: Int? = null,
|
val max_ws_frame_bytes: Int? = null,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* NIP-77 negentropy tuning. Defaults track strfry
|
||||||
|
* (`hoytech/strfry`) so a Geode relay accepts the same workload
|
||||||
|
* shape and exchanges the same NEG-MSG round-trip size as
|
||||||
|
* strfry — the de-facto reference implementation.
|
||||||
|
*
|
||||||
|
* - [frame_size_limit] mirrors strfry's hard-coded
|
||||||
|
* `Negentropy ne(storage, 500'000)` in `RelayNegentropy.cpp`.
|
||||||
|
* Hex-encoded that's ~1 MB on the wire per NEG-MSG; ensure
|
||||||
|
* `[limits].max_ws_frame_bytes` (when set) is at least double
|
||||||
|
* this or NEG-MSGs get truncated by the WS layer.
|
||||||
|
* - [max_sync_events] mirrors strfry's
|
||||||
|
* `relay__negentropy__maxSyncEvents`. NEG-OPEN whose snapshot
|
||||||
|
* exceeds this returns
|
||||||
|
* `["NEG-ERR", "<subId>", "blocked: too many query results"]`.
|
||||||
|
* - [max_sessions_per_connection] caps concurrent NEG-OPEN
|
||||||
|
* sessions held by a single connection. strfry shares its
|
||||||
|
* 200-cap with REQ subs via `relay__maxSubsPerConnection`;
|
||||||
|
* Geode counts NEG independently for now (REQ has no cap yet).
|
||||||
|
* Overflow returns NOTICE
|
||||||
|
* `"too many concurrent NEG requests"` (matches strfry).
|
||||||
|
*/
|
||||||
|
data class NegentropySection(
|
||||||
|
val frame_size_limit: Long = 500_000L,
|
||||||
|
val max_sync_events: Int = 1_000_000,
|
||||||
|
val max_sessions_per_connection: Int = 200,
|
||||||
|
)
|
||||||
|
|
||||||
data class AuthorizationSection(
|
data class AuthorizationSection(
|
||||||
val pubkey_whitelist: List<String> = emptyList(),
|
val pubkey_whitelist: List<String> = emptyList(),
|
||||||
val pubkey_blacklist: List<String> = emptyList(),
|
val pubkey_blacklist: List<String> = emptyList(),
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
|
|||||||
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
||||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
|
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
|
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
|
||||||
@@ -33,6 +34,7 @@ import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
|||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
|
import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage
|
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession
|
||||||
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||||
import kotlinx.coroutines.channels.Channel
|
import kotlinx.coroutines.channels.Channel
|
||||||
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
|
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
|
||||||
import kotlinx.coroutines.runBlocking
|
import kotlinx.coroutines.runBlocking
|
||||||
@@ -237,12 +239,82 @@ class Nip77NegentropyTest {
|
|||||||
val response = client.nextMessage()
|
val response = client.nextMessage()
|
||||||
assertTrue(response is NegErrMessage, "expected NEG-ERR, got ${response::class.simpleName}")
|
assertTrue(response is NegErrMessage, "expected NEG-ERR, got ${response::class.simpleName}")
|
||||||
assertEquals("ghost-sub", response.subId)
|
assertEquals("ghost-sub", response.subId)
|
||||||
assertTrue(response.reason.contains("no negentropy session"))
|
// strfry-parity wording — clients in the wild string-match this.
|
||||||
|
assertEquals("closed: unknown subscription handle", response.reason)
|
||||||
} finally {
|
} finally {
|
||||||
client.close()
|
client.close()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun negOpenSnapshotOverflowReturnsStrFryNegErr() =
|
||||||
|
runBlocking {
|
||||||
|
// Tiny cap so the test is fast. Preload more events than the
|
||||||
|
// cap so NEG-OPEN must reject — strfry's parity behaviour for
|
||||||
|
// `relay__negentropy__maxSyncEvents`.
|
||||||
|
val capped = RelayHub(negentropySettings = NegentropySettings(maxSyncEvents = 5))
|
||||||
|
try {
|
||||||
|
val capUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7771/")
|
||||||
|
val events = makeEvents(20)
|
||||||
|
capped.getOrCreate(capUrl).preload(events)
|
||||||
|
|
||||||
|
val client = WireClient(capped, capUrl)
|
||||||
|
try {
|
||||||
|
val session =
|
||||||
|
NegentropySession(
|
||||||
|
subId = "neg-overflow",
|
||||||
|
filter = Filter(kinds = listOf(1)),
|
||||||
|
localEvents = emptyList(),
|
||||||
|
)
|
||||||
|
client.send(OptimizedJsonMapper.toJson(session.open()))
|
||||||
|
|
||||||
|
val response = client.nextMessage()
|
||||||
|
assertTrue(response is NegErrMessage, "expected NEG-ERR, got ${response::class.simpleName}")
|
||||||
|
assertEquals("neg-overflow", response.subId)
|
||||||
|
// strfry-parity wording.
|
||||||
|
assertEquals("blocked: too many query results", response.reason)
|
||||||
|
} finally {
|
||||||
|
client.close()
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
capped.close()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun negOpenPerConnectionCapEmitsNotice() =
|
||||||
|
runBlocking {
|
||||||
|
// Cap = 2, so the third NEG-OPEN on one connection should
|
||||||
|
// be rejected with a NOTICE (matching strfry's wording).
|
||||||
|
val capped = RelayHub(negentropySettings = NegentropySettings(maxSessionsPerConnection = 2))
|
||||||
|
try {
|
||||||
|
val capUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7772/")
|
||||||
|
capped.getOrCreate(capUrl).preload(makeEvents(3))
|
||||||
|
|
||||||
|
val client = WireClient(capped, capUrl)
|
||||||
|
try {
|
||||||
|
repeat(2) { i ->
|
||||||
|
val s = NegentropySession("ok-$i", Filter(kinds = listOf(1)), localEvents = emptyList())
|
||||||
|
client.send(OptimizedJsonMapper.toJson(s.open()))
|
||||||
|
// Drain the NEG-MSG response so the next OPEN goes
|
||||||
|
// through cleanly.
|
||||||
|
client.nextMessage() as NegMsgMessage
|
||||||
|
}
|
||||||
|
|
||||||
|
// Third OPEN — should be rejected with a NOTICE.
|
||||||
|
val third = NegentropySession("third", Filter(kinds = listOf(1)), localEvents = emptyList())
|
||||||
|
client.send(OptimizedJsonMapper.toJson(third.open()))
|
||||||
|
val response = client.nextMessage()
|
||||||
|
assertTrue(response is NoticeMessage, "expected NOTICE, got ${response::class.simpleName}")
|
||||||
|
assertEquals("too many concurrent NEG requests", response.message)
|
||||||
|
} finally {
|
||||||
|
client.close()
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
capped.close()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun negOpenWithSameSubIdReplacesPriorSession() =
|
fun negOpenWithSameSubIdReplacesPriorSession() =
|
||||||
runBlocking {
|
runBlocking {
|
||||||
|
|||||||
+6
@@ -24,6 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
|
|||||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Decorator that canonicalises every [Event] returned by the inner
|
* Decorator that canonicalises every [Event] returned by the inner
|
||||||
@@ -111,6 +112,11 @@ class InterningEventStore(
|
|||||||
|
|
||||||
override suspend fun count(filters: List<Filter>): Int = inner.count(filters)
|
override suspend fun count(filters: List<Filter>): Int = inner.count(filters)
|
||||||
|
|
||||||
|
override suspend fun snapshotIdsForNegentropy(
|
||||||
|
filters: List<Filter>,
|
||||||
|
maxEntries: Int?,
|
||||||
|
): List<IdAndTime> = inner.snapshotIdsForNegentropy(filters, maxEntries)
|
||||||
|
|
||||||
override suspend fun delete(filter: Filter) = inner.delete(filter)
|
override suspend fun delete(filter: Filter) = inner.delete(filter)
|
||||||
|
|
||||||
override suspend fun delete(filters: List<Filter>) = inner.delete(filters)
|
override suspend fun delete(filters: List<Filter>) = inner.delete(filters)
|
||||||
|
|||||||
+16
@@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.server
|
|||||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||||
import kotlinx.coroutines.CompletableDeferred
|
import kotlinx.coroutines.CompletableDeferred
|
||||||
import kotlinx.coroutines.channels.BufferOverflow
|
import kotlinx.coroutines.channels.BufferOverflow
|
||||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||||
@@ -161,4 +162,19 @@ class LiveEventStore(
|
|||||||
}
|
}
|
||||||
return merged
|
return merged
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Lightweight snapshot for NIP-77 negentropy. Returns
|
||||||
|
* `(created_at, id)` pairs only — no Event materialisation —
|
||||||
|
* matching strfry's `MemoryView` footprint of ~40 B/entry.
|
||||||
|
*
|
||||||
|
* If [maxEntries] is non-null, the underlying store may return
|
||||||
|
* up to `maxEntries + 1` entries; the +1 sentinel lets the
|
||||||
|
* caller distinguish "exactly at cap" from "exceeds cap" without
|
||||||
|
* scanning past the cap.
|
||||||
|
*/
|
||||||
|
suspend fun snapshotIdsForNegentropy(
|
||||||
|
filters: List<Filter>,
|
||||||
|
maxEntries: Int? = null,
|
||||||
|
): List<IdAndTime> = store.snapshotIdsForNegentropy(filters, maxEntries)
|
||||||
}
|
}
|
||||||
|
|||||||
+54
-8
@@ -21,12 +21,14 @@
|
|||||||
package com.vitorpamplona.quartz.nip01Core.relay.server
|
package com.vitorpamplona.quartz.nip01Core.relay.server
|
||||||
|
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
|
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
|
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd
|
import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
|
import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd
|
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession
|
||||||
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Per-connection NIP-77 negentropy state and dispatch.
|
* Per-connection NIP-77 negentropy state and dispatch.
|
||||||
@@ -39,21 +41,37 @@ import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession
|
|||||||
* Plain [HashMap] is sufficient because the registry is mutated only
|
* Plain [HashMap] is sufficient because the registry is mutated only
|
||||||
* from [RelaySession.receive] — that path is single-threaded per the
|
* from [RelaySession.receive] — that path is single-threaded per the
|
||||||
* WebSocket handler contract.
|
* WebSocket handler contract.
|
||||||
|
*
|
||||||
|
* Defaults match strfry (`hoytech/strfry`) so a Geode relay reconciles
|
||||||
|
* with the same round-trip shape and the same operator-visible
|
||||||
|
* protections — see [NegentropySettings].
|
||||||
*/
|
*/
|
||||||
class NegSessionRegistry(
|
class NegSessionRegistry(
|
||||||
private val store: LiveEventStore,
|
private val store: LiveEventStore,
|
||||||
private val send: (Message) -> Unit,
|
private val send: (Message) -> Unit,
|
||||||
|
private val settings: NegentropySettings = NegentropySettings.Default,
|
||||||
) {
|
) {
|
||||||
private val sessions = HashMap<String, NegentropyServerSession>()
|
private val sessions = HashMap<String, NegentropyServerSession>()
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Open a reconciliation session. The relay snapshots its matching
|
* Open a reconciliation session. The relay snapshots the matching
|
||||||
* events at this instant — concurrent inserts during the sync are
|
* `(created_at, id)` pairs at this instant — concurrent inserts
|
||||||
* not surfaced; clients re-open if they want fresh state.
|
* during the sync are not surfaced; clients re-open if they want
|
||||||
|
* fresh state.
|
||||||
*
|
*
|
||||||
* Access control reuses the REQ policy hook: a relay that requires
|
* Access control reuses the REQ policy hook: a relay that requires
|
||||||
* AUTH or has kind/pubkey allow-deny lists applies the same rules
|
* AUTH or has kind/pubkey allow-deny lists applies the same rules
|
||||||
* to NEG-OPEN as it does to subscription REQs.
|
* to NEG-OPEN as it does to subscription REQs.
|
||||||
|
*
|
||||||
|
* Two strfry-parity protections fire here:
|
||||||
|
* - **Per-connection session cap.** If an OPEN would push the
|
||||||
|
* map past [NegentropySettings.maxSessionsPerConnection], we
|
||||||
|
* send a NOTICE (matching strfry's
|
||||||
|
* `"too many concurrent NEG requests"`) and drop the OPEN.
|
||||||
|
* - **Snapshot size cap.** The store is asked for at most
|
||||||
|
* `maxSyncEvents + 1` entries; if the +1 sentinel comes back,
|
||||||
|
* the corpus exceeds the cap and we send NEG-ERR
|
||||||
|
* `"blocked: too many query results"` (matching strfry).
|
||||||
*/
|
*/
|
||||||
suspend fun open(
|
suspend fun open(
|
||||||
cmd: NegOpenCmd,
|
cmd: NegOpenCmd,
|
||||||
@@ -66,20 +84,43 @@ class NegSessionRegistry(
|
|||||||
}
|
}
|
||||||
val filters = (gate as PolicyResult.Accepted).cmd.filters
|
val filters = (gate as PolicyResult.Accepted).cmd.filters
|
||||||
|
|
||||||
|
// Per-connection cap. Only fires when this is a NEW subId —
|
||||||
|
// a same-subId re-open replaces the prior session 1-for-1.
|
||||||
|
val isReopen = sessions.containsKey(cmd.subId)
|
||||||
|
if (!isReopen && sessions.size >= settings.maxSessionsPerConnection) {
|
||||||
|
send(NoticeMessage("too many concurrent NEG requests"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// NIP-77: same-subId OPEN replaces any prior session.
|
// NIP-77: same-subId OPEN replaces any prior session.
|
||||||
sessions.remove(cmd.subId)
|
sessions.remove(cmd.subId)
|
||||||
|
|
||||||
val events = store.snapshotQuery(filters)
|
val cap = settings.maxSyncEvents
|
||||||
val session = NegentropyServerSession(cmd.subId, events)
|
val entries = store.snapshotIdsForNegentropy(filters, maxEntries = cap)
|
||||||
|
if (entries.size > cap) {
|
||||||
|
send(NegErrMessage(cmd.subId, "blocked: too many query results"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
val session =
|
||||||
|
NegentropyServerSession(
|
||||||
|
subId = cmd.subId,
|
||||||
|
localEntries = entries,
|
||||||
|
frameSizeLimit = settings.frameSizeLimit,
|
||||||
|
)
|
||||||
sessions[cmd.subId] = session
|
sessions[cmd.subId] = session
|
||||||
|
|
||||||
runMessage(cmd.subId, session) { it.processMessage(cmd.initialMessage) }
|
runMessage(cmd.subId, session) { it.processMessage(cmd.initialMessage) }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Process a follow-up NEG-MSG. strfry-parity wording for the
|
||||||
|
* unknown-subId case: `"closed: unknown subscription handle"`.
|
||||||
|
*/
|
||||||
fun msg(cmd: NegMsgCmd) {
|
fun msg(cmd: NegMsgCmd) {
|
||||||
val session = sessions[cmd.subId]
|
val session = sessions[cmd.subId]
|
||||||
if (session == null) {
|
if (session == null) {
|
||||||
send(NegErrMessage(cmd.subId, "error: no negentropy session for ${cmd.subId}"))
|
send(NegErrMessage(cmd.subId, "closed: unknown subscription handle"))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
runMessage(cmd.subId, session) { it.processMessage(cmd.message) }
|
runMessage(cmd.subId, session) { it.processMessage(cmd.message) }
|
||||||
@@ -99,6 +140,9 @@ class NegSessionRegistry(
|
|||||||
sessions.clear()
|
sessions.clear()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Test/diagnostic accessor. */
|
||||||
|
val activeSessionCount: Int get() = sessions.size
|
||||||
|
|
||||||
private inline fun runMessage(
|
private inline fun runMessage(
|
||||||
subId: String,
|
subId: String,
|
||||||
session: NegentropyServerSession,
|
session: NegentropyServerSession,
|
||||||
@@ -107,9 +151,11 @@ class NegSessionRegistry(
|
|||||||
try {
|
try {
|
||||||
val response = block(session)
|
val response = block(session)
|
||||||
if (response != null) send(response)
|
if (response != null) send(response)
|
||||||
} catch (e: Exception) {
|
} catch (_: Exception) {
|
||||||
|
// strfry sends `PROTOCOL-ERROR` on library reconcile()
|
||||||
|
// parse failure and tears the session down.
|
||||||
sessions.remove(subId)
|
sessions.remove(subId)
|
||||||
send(NegErrMessage(subId, "error: ${e.message ?: e::class.simpleName}"))
|
send(NegErrMessage(subId, "PROTOCOL-ERROR"))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+6
@@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.server
|
|||||||
import com.vitorpamplona.quartz.nip01Core.crypto.verify
|
import com.vitorpamplona.quartz.nip01Core.crypto.verify
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy
|
import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||||
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||||
import com.vitorpamplona.quartz.utils.cache.LargeCache
|
import com.vitorpamplona.quartz.utils.cache.LargeCache
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
import kotlinx.coroutines.SupervisorJob
|
import kotlinx.coroutines.SupervisorJob
|
||||||
@@ -43,12 +44,16 @@ import kotlin.coroutines.CoroutineContext
|
|||||||
* coroutine inside [VerifyPolicy]. Callers that flip this on should
|
* coroutine inside [VerifyPolicy]. Callers that flip this on should
|
||||||
* *omit* `VerifyPolicy` from their [policyBuilder] chain to avoid
|
* *omit* `VerifyPolicy` from their [policyBuilder] chain to avoid
|
||||||
* double-verifying.
|
* double-verifying.
|
||||||
|
* @param negentropySettings NIP-77 server-side tuning (frame cap,
|
||||||
|
* snapshot cap, per-connection session cap). Defaults to strfry-
|
||||||
|
* parity values; see [NegentropySettings].
|
||||||
*/
|
*/
|
||||||
class NostrServer(
|
class NostrServer(
|
||||||
private val store: IEventStore,
|
private val store: IEventStore,
|
||||||
private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy },
|
private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy },
|
||||||
private val parentContext: CoroutineContext = SupervisorJob(),
|
private val parentContext: CoroutineContext = SupervisorJob(),
|
||||||
parallelVerify: Boolean = false,
|
parallelVerify: Boolean = false,
|
||||||
|
private val negentropySettings: NegentropySettings = NegentropySettings.Default,
|
||||||
) : AutoCloseable {
|
) : AutoCloseable {
|
||||||
/** Scope for all subscriptions. */
|
/** Scope for all subscriptions. */
|
||||||
private val scope = CoroutineScope(parentContext + SupervisorJob())
|
private val scope = CoroutineScope(parentContext + SupervisorJob())
|
||||||
@@ -87,6 +92,7 @@ class NostrServer(
|
|||||||
onClose = { session ->
|
onClose = { session ->
|
||||||
connections.remove(session.hashCode())
|
connections.remove(session.hashCode())
|
||||||
},
|
},
|
||||||
|
negentropySettings = negentropySettings,
|
||||||
).also { session ->
|
).also { session ->
|
||||||
connections.put(session.hashCode(), session)
|
connections.put(session.hashCode(), session)
|
||||||
}
|
}
|
||||||
|
|||||||
+3
-1
@@ -38,6 +38,7 @@ import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
|||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd
|
import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd
|
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd
|
||||||
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
||||||
|
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||||
import com.vitorpamplona.quartz.utils.Log
|
import com.vitorpamplona.quartz.utils.Log
|
||||||
import com.vitorpamplona.quartz.utils.cache.LargeCache
|
import com.vitorpamplona.quartz.utils.cache.LargeCache
|
||||||
import kotlinx.coroutines.CancellationException
|
import kotlinx.coroutines.CancellationException
|
||||||
@@ -56,11 +57,12 @@ class RelaySession(
|
|||||||
private val scope: CoroutineScope,
|
private val scope: CoroutineScope,
|
||||||
private val onSend: (String) -> Unit,
|
private val onSend: (String) -> Unit,
|
||||||
private val onClose: (RelaySession) -> Unit,
|
private val onClose: (RelaySession) -> Unit,
|
||||||
|
negentropySettings: NegentropySettings = NegentropySettings.Default,
|
||||||
) : AutoCloseable {
|
) : AutoCloseable {
|
||||||
private val subscriptions = LargeCache<String, Job>()
|
private val subscriptions = LargeCache<String, Job>()
|
||||||
|
|
||||||
/** NIP-77 negentropy state for this connection. */
|
/** NIP-77 negentropy state for this connection. */
|
||||||
private val negentropy = NegSessionRegistry(store, ::send)
|
private val negentropy = NegSessionRegistry(store, ::send, negentropySettings)
|
||||||
|
|
||||||
private fun addSubscription(
|
private fun addSubscription(
|
||||||
subId: String,
|
subId: String,
|
||||||
|
|||||||
@@ -95,6 +95,38 @@ interface IEventStore : AutoCloseable {
|
|||||||
|
|
||||||
suspend fun count(filters: List<Filter>): Int
|
suspend fun count(filters: List<Filter>): Int
|
||||||
|
|
||||||
|
/**
|
||||||
|
* NIP-77 negentropy snapshot. Returns `(created_at, id)` pairs
|
||||||
|
* for every event matching [filters], with no content/tags/sig
|
||||||
|
* decode. Used by the server-side reconciliation path to build a
|
||||||
|
* `StorageVector` without materialising full [Event] objects —
|
||||||
|
* ~40 B/entry instead of ~1 KB/entry. Order is unspecified;
|
||||||
|
* negentropy's `seal()` re-sorts.
|
||||||
|
*
|
||||||
|
* If [maxEntries] is non-null, the implementation may return up
|
||||||
|
* to `maxEntries + 1` entries; the caller compares the result
|
||||||
|
* size to detect overflow (matching strfry's `maxSyncEvents`
|
||||||
|
* guard). The +1 sentinel lets the caller distinguish "exactly
|
||||||
|
* capped" from "too many to fit".
|
||||||
|
*
|
||||||
|
* Default implementation falls back to the full-decode path so
|
||||||
|
* non-SQLite stores stay correct; SQLite overrides with a direct
|
||||||
|
* `SELECT id, created_at` against the `query_by_created_at_id`
|
||||||
|
* index. Honors the same filter semantics as [query] including
|
||||||
|
* any `limit`.
|
||||||
|
*/
|
||||||
|
suspend fun snapshotIdsForNegentropy(
|
||||||
|
filters: List<Filter>,
|
||||||
|
maxEntries: Int? = null,
|
||||||
|
): List<IdAndTime> {
|
||||||
|
val all = query<Event>(filters).map { IdAndTime(it.createdAt, it.id) }
|
||||||
|
return if (maxEntries != null && all.size > maxEntries + 1) {
|
||||||
|
all.subList(0, maxEntries + 1)
|
||||||
|
} else {
|
||||||
|
all
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
suspend fun delete(filter: Filter)
|
suspend fun delete(filter: Filter)
|
||||||
|
|
||||||
suspend fun delete(filters: List<Filter>)
|
suspend fun delete(filters: List<Filter>)
|
||||||
|
|||||||
@@ -0,0 +1,39 @@
|
|||||||
|
/*
|
||||||
|
* 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
|
||||||
|
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Lightweight projection of an event used by NIP-77 negentropy: just
|
||||||
|
* the two fields the reconciliation library indexes — `created_at`
|
||||||
|
* and the 32-byte event id.
|
||||||
|
*
|
||||||
|
* Returned by [IEventStore.snapshotIdsForNegentropy] so the relay can
|
||||||
|
* build a [com.vitorpamplona.negentropy.storage.StorageVector] without
|
||||||
|
* materialising full [com.vitorpamplona.quartz.nip01Core.core.Event]
|
||||||
|
* objects (content, tags, sig). For a 1 M-event snapshot this drops
|
||||||
|
* peak heap from ~1 GB to ~40 MB — strfry's `MemoryView` parity.
|
||||||
|
*/
|
||||||
|
data class IdAndTime(
|
||||||
|
val createdAt: Long,
|
||||||
|
val id: HexKey,
|
||||||
|
)
|
||||||
+5
@@ -189,6 +189,11 @@ class ObservableEventStore(
|
|||||||
|
|
||||||
override suspend fun count(filters: List<Filter>): Int = inner.count(filters)
|
override suspend fun count(filters: List<Filter>): Int = inner.count(filters)
|
||||||
|
|
||||||
|
override suspend fun snapshotIdsForNegentropy(
|
||||||
|
filters: List<Filter>,
|
||||||
|
maxEntries: Int?,
|
||||||
|
): List<IdAndTime> = inner.snapshotIdsForNegentropy(filters, maxEntries)
|
||||||
|
|
||||||
override suspend fun delete(filter: Filter) {
|
override suspend fun delete(filter: Filter) {
|
||||||
inner.delete(filter)
|
inner.delete(filter)
|
||||||
_changes.emit(StoreChange.DeleteByFilter(listOf(filter)))
|
_changes.emit(StoreChange.DeleteByFilter(listOf(filter)))
|
||||||
|
|||||||
+6
@@ -26,6 +26,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
|||||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
|
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* SQLite-backed [IEventStore] with default DB-file name and relay
|
* SQLite-backed [IEventStore] with default DB-file name and relay
|
||||||
@@ -63,6 +64,11 @@ class EventStore(
|
|||||||
|
|
||||||
override suspend fun count(filters: List<Filter>) = store.count(filters)
|
override suspend fun count(filters: List<Filter>) = store.count(filters)
|
||||||
|
|
||||||
|
override suspend fun snapshotIdsForNegentropy(
|
||||||
|
filters: List<Filter>,
|
||||||
|
maxEntries: Int?,
|
||||||
|
): List<IdAndTime> = store.snapshotIdsForNegentropy(filters, maxEntries)
|
||||||
|
|
||||||
override suspend fun delete(filter: Filter) {
|
override suspend fun delete(filter: Filter) {
|
||||||
store.delete(filter)
|
store.delete(filter)
|
||||||
}
|
}
|
||||||
|
|||||||
+217
@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Kind
|
|||||||
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
||||||
import com.vitorpamplona.quartz.nip01Core.core.isAddressable
|
import com.vitorpamplona.quartz.nip01Core.core.isAddressable
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.sql.where
|
import com.vitorpamplona.quartz.nip01Core.store.sqlite.sql.where
|
||||||
import com.vitorpamplona.quartz.utils.EventFactory
|
import com.vitorpamplona.quartz.utils.EventFactory
|
||||||
|
|
||||||
@@ -172,6 +173,222 @@ class QueryBuilder(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// -----------------------------------------------------------------
|
||||||
|
// NIP-77 negentropy snapshot path
|
||||||
|
//
|
||||||
|
// Projects only (id, created_at) — no content/tags/sig decode —
|
||||||
|
// so the relay can build a StorageVector without materialising
|
||||||
|
// full Event objects. ~40 B/entry instead of ~1 KB/entry.
|
||||||
|
// No ORDER BY: negentropy's seal() re-sorts. No limit injection:
|
||||||
|
// the per-session cap is enforced upstream as a count check.
|
||||||
|
// -----------------------------------------------------------------
|
||||||
|
fun snapshotIdsForNegentropy(
|
||||||
|
filters: List<Filter>,
|
||||||
|
db: SQLiteConnection,
|
||||||
|
maxEntries: Int? = null,
|
||||||
|
): List<IdAndTime> {
|
||||||
|
val inner =
|
||||||
|
if (filters.size == 1) {
|
||||||
|
toSnapshotIdsSql(filters.first(), hasher(db))
|
||||||
|
} else {
|
||||||
|
toSnapshotIdsSql(filters, hasher(db))
|
||||||
|
}
|
||||||
|
// Safety cap: wrap with `LIMIT maxEntries + 1` so we can
|
||||||
|
// detect overflow without scanning beyond the cap. The +1
|
||||||
|
// sentinel lets the caller distinguish "exactly capped" from
|
||||||
|
// "too many to fit". Matches strfry's `maxSyncEvents` guard.
|
||||||
|
val query =
|
||||||
|
if (maxEntries != null) {
|
||||||
|
QuerySpec(
|
||||||
|
"SELECT id, created_at FROM (${inner.sql}) LIMIT ${maxEntries + 1}",
|
||||||
|
inner.args,
|
||||||
|
)
|
||||||
|
} else {
|
||||||
|
inner
|
||||||
|
}
|
||||||
|
return db.runIdAndTimeQuery(query)
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun toSnapshotIdsSql(
|
||||||
|
filter: Filter,
|
||||||
|
hasher: TagNameValueHasher,
|
||||||
|
): QuerySpec {
|
||||||
|
val newFilter = filter.toFilterWithDTags()
|
||||||
|
|
||||||
|
// Simple path — no tag joins, no FTS — collapses to a single
|
||||||
|
// SELECT against event_headers.
|
||||||
|
if (newFilter.isSimpleQuery()) {
|
||||||
|
return makeSimpleIdsQuery(
|
||||||
|
ids = newFilter.ids,
|
||||||
|
authors = newFilter.authors,
|
||||||
|
kinds = newFilter.kinds,
|
||||||
|
dTags = newFilter.dTags,
|
||||||
|
since = newFilter.since,
|
||||||
|
until = newFilter.until,
|
||||||
|
limit = newFilter.limit,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Search path — FTS join. Project id+created_at off
|
||||||
|
// event_headers via the FTS row_id linkage.
|
||||||
|
if (newFilter.isSimpleSearch()) {
|
||||||
|
return makeSimpleIdsSearch(
|
||||||
|
search = newFilter.search!!,
|
||||||
|
ids = newFilter.ids,
|
||||||
|
authors = newFilter.authors,
|
||||||
|
kinds = newFilter.kinds,
|
||||||
|
dTags = newFilter.dTags,
|
||||||
|
since = newFilter.since,
|
||||||
|
until = newFilter.until,
|
||||||
|
limit = newFilter.limit,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Tag-join path — reuse the existing row_id subquery and
|
||||||
|
// join back to event_headers for the projection.
|
||||||
|
val rowIdSubquery = prepareRowIDSubQueries(filter, hasher)
|
||||||
|
return if (rowIdSubquery == null) {
|
||||||
|
QuerySpec("SELECT id, created_at FROM event_headers")
|
||||||
|
} else {
|
||||||
|
QuerySpec(
|
||||||
|
"""
|
||||||
|
SELECT event_headers.id, event_headers.created_at FROM event_headers
|
||||||
|
INNER JOIN (
|
||||||
|
${rowIdSubquery.sql}
|
||||||
|
) AS filtered
|
||||||
|
ON event_headers.row_id = filtered.row_id
|
||||||
|
""".trimIndent(),
|
||||||
|
rowIdSubquery.args,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun toSnapshotIdsSql(
|
||||||
|
filters: List<Filter>,
|
||||||
|
hasher: TagNameValueHasher,
|
||||||
|
): QuerySpec {
|
||||||
|
val rowIdSubqueries = unionSubqueriesIfNeeded(filters, hasher)
|
||||||
|
return if (rowIdSubqueries == null) {
|
||||||
|
QuerySpec("SELECT id, created_at FROM event_headers")
|
||||||
|
} else {
|
||||||
|
QuerySpec(
|
||||||
|
"""
|
||||||
|
SELECT DISTINCT event_headers.id, event_headers.created_at FROM event_headers
|
||||||
|
INNER JOIN (
|
||||||
|
${rowIdSubqueries.sql}
|
||||||
|
) AS filtered
|
||||||
|
ON event_headers.row_id = filtered.row_id
|
||||||
|
""".trimIndent(),
|
||||||
|
rowIdSubqueries.args,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun makeSimpleIdsQuery(
|
||||||
|
ids: List<HexKey>? = null,
|
||||||
|
authors: List<HexKey>? = null,
|
||||||
|
kinds: List<Kind>? = null,
|
||||||
|
dTags: List<String>? = null,
|
||||||
|
since: Long? = null,
|
||||||
|
until: Long? = null,
|
||||||
|
limit: Int? = null,
|
||||||
|
): QuerySpec {
|
||||||
|
val clause =
|
||||||
|
where {
|
||||||
|
ids?.let { equalsOrIn("id", it) }
|
||||||
|
kinds?.let { equalsOrIn("kind", it) }
|
||||||
|
authors?.let { equalsOrIn("pubkey", it) }
|
||||||
|
dTags?.let { equalsOrIn("d_tag", it) }
|
||||||
|
since?.let { greaterThanOrEquals("created_at", it) }
|
||||||
|
until?.let { lessThanOrEquals("created_at", it) }
|
||||||
|
if (dTags != null && kinds != null) {
|
||||||
|
if (kinds.all { it.isAddressable() }) {
|
||||||
|
raw("(kind >= 30000 AND kind < 40000)")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
val sql =
|
||||||
|
buildString {
|
||||||
|
append("SELECT id, created_at FROM event_headers")
|
||||||
|
if (clause.conditions.isNotEmpty()) {
|
||||||
|
append("\nWHERE ")
|
||||||
|
append(clause.conditions)
|
||||||
|
}
|
||||||
|
// Negentropy honors filter `limit` like REQ does
|
||||||
|
// (matches strfry's NostrFilterGroup behaviour).
|
||||||
|
// ORDER BY is required for LIMIT to be meaningful.
|
||||||
|
if (limit != null) {
|
||||||
|
append("\nORDER BY created_at DESC")
|
||||||
|
if (indexStrategy.useAndIndexIdOnOrderBy) {
|
||||||
|
append(", id ASC")
|
||||||
|
}
|
||||||
|
append("\nLIMIT ")
|
||||||
|
append(limit)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return QuerySpec(sql, clause.args)
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun makeSimpleIdsSearch(
|
||||||
|
search: String,
|
||||||
|
ids: List<HexKey>? = null,
|
||||||
|
authors: List<HexKey>? = null,
|
||||||
|
kinds: List<Kind>? = null,
|
||||||
|
dTags: List<String>? = null,
|
||||||
|
since: Long? = null,
|
||||||
|
until: Long? = null,
|
||||||
|
limit: Int? = null,
|
||||||
|
): QuerySpec {
|
||||||
|
val clause =
|
||||||
|
where {
|
||||||
|
ids?.let { equalsOrIn("event_headers.id", it) }
|
||||||
|
match(fts.tableName, search)
|
||||||
|
kinds?.let { equalsOrIn("event_headers.kind", it) }
|
||||||
|
authors?.let { equalsOrIn("event_headers.pubkey", it) }
|
||||||
|
dTags?.let { equalsOrIn("event_headers.d_tag", it) }
|
||||||
|
since?.let { greaterThanOrEquals("event_headers.created_at", it) }
|
||||||
|
until?.let { lessThanOrEquals("event_headers.created_at", it) }
|
||||||
|
if (dTags != null && kinds != null) {
|
||||||
|
if (kinds.all { it.isAddressable() }) {
|
||||||
|
raw("(event_headers.kind >= 30000 AND kind < 40000)")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
val sql =
|
||||||
|
buildString {
|
||||||
|
append("SELECT event_headers.id, event_headers.created_at FROM event_headers")
|
||||||
|
append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.${fts.eventHeaderRowIdName}")
|
||||||
|
if (clause.conditions.isNotEmpty()) {
|
||||||
|
append("\nWHERE ${clause.conditions}")
|
||||||
|
}
|
||||||
|
if (limit != null) {
|
||||||
|
append("\nORDER BY event_headers.created_at DESC")
|
||||||
|
if (indexStrategy.useAndIndexIdOnOrderBy) {
|
||||||
|
append(", event_headers.id ASC")
|
||||||
|
}
|
||||||
|
append("\nLIMIT ")
|
||||||
|
append(limit)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return QuerySpec(sql, clause.args)
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun SQLiteConnection.runIdAndTimeQuery(query: QuerySpec): List<IdAndTime> =
|
||||||
|
prepare(query.sql).use { stmt ->
|
||||||
|
query.args.forEachIndexed { index, arg ->
|
||||||
|
stmt.bindText(index + 1, arg)
|
||||||
|
}
|
||||||
|
val results = ArrayList<IdAndTime>()
|
||||||
|
while (stmt.step()) {
|
||||||
|
results.add(IdAndTime(stmt.getLong(1), stmt.getText(0)))
|
||||||
|
}
|
||||||
|
results
|
||||||
|
}
|
||||||
|
|
||||||
private fun makeEverythingQuery() = "SELECT id, pubkey, created_at, kind, tags, content, sig FROM event_headers ORDER BY created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}"
|
private fun makeEverythingQuery() = "SELECT id, pubkey, created_at, kind, tags, content, sig FROM event_headers ORDER BY created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}"
|
||||||
|
|
||||||
private fun makeQueryIn(rowIdQuery: String) =
|
private fun makeQueryIn(rowIdQuery: String) =
|
||||||
|
|||||||
+6
@@ -32,6 +32,7 @@ import com.vitorpamplona.quartz.nip01Core.core.isEphemeral
|
|||||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||||
import com.vitorpamplona.quartz.nip40Expiration.isExpired
|
import com.vitorpamplona.quartz.nip40Expiration.isExpired
|
||||||
import com.vitorpamplona.quartz.utils.EventFactory
|
import com.vitorpamplona.quartz.utils.EventFactory
|
||||||
|
|
||||||
@@ -315,6 +316,11 @@ class SQLiteEventStore(
|
|||||||
|
|
||||||
suspend fun count(filters: List<Filter>): Int = pool.useReader { queryBuilder.count(filters, it) }
|
suspend fun count(filters: List<Filter>): Int = pool.useReader { queryBuilder.count(filters, it) }
|
||||||
|
|
||||||
|
suspend fun snapshotIdsForNegentropy(
|
||||||
|
filters: List<Filter>,
|
||||||
|
maxEntries: Int? = null,
|
||||||
|
): List<IdAndTime> = pool.useReader { queryBuilder.snapshotIdsForNegentropy(filters, it, maxEntries) }
|
||||||
|
|
||||||
suspend fun delete(filter: Filter) = pool.useWriter { queryBuilder.delete(filter, it) }
|
suspend fun delete(filter: Filter) = pool.useWriter { queryBuilder.delete(filter, it) }
|
||||||
|
|
||||||
suspend fun delete(filters: List<Filter>) = pool.useWriter { queryBuilder.delete(filters, it) }
|
suspend fun delete(filters: List<Filter>) = pool.useWriter { queryBuilder.delete(filters, it) }
|
||||||
|
|||||||
+43
-5
@@ -23,6 +23,7 @@ package com.vitorpamplona.quartz.nip77Negentropy
|
|||||||
import com.vitorpamplona.negentropy.Negentropy
|
import com.vitorpamplona.negentropy.Negentropy
|
||||||
import com.vitorpamplona.negentropy.storage.StorageVector
|
import com.vitorpamplona.negentropy.storage.StorageVector
|
||||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||||
import com.vitorpamplona.quartz.utils.Hex
|
import com.vitorpamplona.quartz.utils.Hex
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -31,28 +32,65 @@ import com.vitorpamplona.quartz.utils.Hex
|
|||||||
* Used when acting as a relay (or relay-relay sync) to respond to
|
* Used when acting as a relay (or relay-relay sync) to respond to
|
||||||
* incoming NEG-OPEN and NEG-MSG from a client.
|
* incoming NEG-OPEN and NEG-MSG from a client.
|
||||||
*
|
*
|
||||||
|
* The constructor takes [IdAndTime] entries (just `created_at` and the
|
||||||
|
* 32-byte event id) to keep the per-session footprint at ~40 B/entry —
|
||||||
|
* matching strfry's `MemoryView` path. A [List]<Event> overload is
|
||||||
|
* kept for callers (and tests) that already hold full events.
|
||||||
|
*
|
||||||
* Usage:
|
* Usage:
|
||||||
* 1. On NEG-OPEN: create a [NegentropyServerSession] with the matching local events
|
* 1. On NEG-OPEN: create a [NegentropyServerSession] with the matching local entries
|
||||||
* 2. Call [processMessage] with the initial hex message from NEG-OPEN
|
* 2. Call [processMessage] with the initial hex message from NEG-OPEN
|
||||||
* 3. Send back the resulting [NegMsgMessage]
|
* 3. Send back the resulting [NegMsgMessage]
|
||||||
* 4. On subsequent NEG-MSG: call [processMessage] again and send the response
|
* 4. On subsequent NEG-MSG: call [processMessage] again and send the response
|
||||||
|
*
|
||||||
|
* @param frameSizeLimit max bytes per NEG-MSG response (raw payload,
|
||||||
|
* before hex). Default `500_000` matches strfry's hard-coded
|
||||||
|
* `Negentropy ne(storage, 500'000)` so a single round-trip carries
|
||||||
|
* the same payload as strfry's reconciliation.
|
||||||
*/
|
*/
|
||||||
class NegentropyServerSession(
|
class NegentropyServerSession(
|
||||||
val subId: String,
|
val subId: String,
|
||||||
localEvents: List<Event>,
|
localEntries: List<IdAndTime>,
|
||||||
frameSizeLimit: Long = 0,
|
frameSizeLimit: Long = DEFAULT_FRAME_SIZE_LIMIT,
|
||||||
) {
|
) {
|
||||||
private val storage = StorageVector()
|
private val storage = StorageVector()
|
||||||
private val negentropy: Negentropy
|
private val negentropy: Negentropy
|
||||||
|
|
||||||
init {
|
init {
|
||||||
for (event in localEvents) {
|
for (entry in localEntries) {
|
||||||
storage.insert(event.createdAt, event.id)
|
storage.insert(entry.createdAt, entry.id)
|
||||||
}
|
}
|
||||||
storage.seal()
|
storage.seal()
|
||||||
negentropy = Negentropy(storage, frameSizeLimit)
|
negentropy = Negentropy(storage, frameSizeLimit)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
companion object {
|
||||||
|
/**
|
||||||
|
* strfry parity: `Negentropy ne(storage, 500'000)` in
|
||||||
|
* `RelayNegentropy.cpp`. Hex-encoded that's ~1 MB on the wire
|
||||||
|
* per NEG-MSG, the de-facto sync round-trip size.
|
||||||
|
*/
|
||||||
|
const val DEFAULT_FRAME_SIZE_LIMIT: Long = 500_000L
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Convenience for callers that hold full [Event] objects
|
||||||
|
* (mostly tests + relay-relay sync paths). Production server
|
||||||
|
* code should call the [IdAndTime] constructor directly via
|
||||||
|
* `IEventStore.snapshotIdsForNegentropy` to avoid the full
|
||||||
|
* Event materialisation that this projection collapses.
|
||||||
|
*/
|
||||||
|
fun fromEvents(
|
||||||
|
subId: String,
|
||||||
|
localEvents: List<Event>,
|
||||||
|
frameSizeLimit: Long = DEFAULT_FRAME_SIZE_LIMIT,
|
||||||
|
): NegentropyServerSession =
|
||||||
|
NegentropyServerSession(
|
||||||
|
subId = subId,
|
||||||
|
localEntries = localEvents.map { IdAndTime(it.createdAt, it.id) },
|
||||||
|
frameSizeLimit = frameSizeLimit,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
fun processMessage(hexMessage: String): NegMsgMessage? {
|
fun processMessage(hexMessage: String): NegMsgMessage? {
|
||||||
val msgBytes = Hex.decode(hexMessage)
|
val msgBytes = Hex.decode(hexMessage)
|
||||||
val result = negentropy.reconcile(msgBytes)
|
val result = negentropy.reconcile(msgBytes)
|
||||||
|
|||||||
+50
@@ -0,0 +1,50 @@
|
|||||||
|
/*
|
||||||
|
* 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
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Server-side NIP-77 tuning. Defaults track strfry
|
||||||
|
* (`hoytech/strfry`) so a Quartz-based relay accepts the same
|
||||||
|
* workload shape and exchanges the same NEG-MSG round-trip size.
|
||||||
|
*
|
||||||
|
* @param frameSizeLimit Max bytes per NEG-MSG response payload
|
||||||
|
* (raw, before hex). 500_000 matches strfry's hard-coded
|
||||||
|
* `Negentropy ne(storage, 500'000)` in `RelayNegentropy.cpp`.
|
||||||
|
* The `kmp-negentropy` library enforces `>= 4096` (or `0` for
|
||||||
|
* unlimited).
|
||||||
|
* @param maxSyncEvents Hard cap on the snapshot size for a single
|
||||||
|
* NEG-OPEN. Mirrors strfry's `relay__negentropy__maxSyncEvents`.
|
||||||
|
* Overflow returns NEG-ERR `"blocked: too many query results"`.
|
||||||
|
* @param maxSessionsPerConnection Cap on concurrent NEG sessions
|
||||||
|
* held by one connection. strfry shares 200 with REQ subs; we
|
||||||
|
* count NEG independently. Overflow sends NOTICE
|
||||||
|
* `"too many concurrent NEG requests"`.
|
||||||
|
*/
|
||||||
|
data class NegentropySettings(
|
||||||
|
val frameSizeLimit: Long = NegentropyServerSession.DEFAULT_FRAME_SIZE_LIMIT,
|
||||||
|
val maxSyncEvents: Int = 1_000_000,
|
||||||
|
val maxSessionsPerConnection: Int = 200,
|
||||||
|
) {
|
||||||
|
companion object {
|
||||||
|
/** strfry-equivalent defaults. */
|
||||||
|
val Default = NegentropySettings()
|
||||||
|
}
|
||||||
|
}
|
||||||
+103
@@ -0,0 +1,103 @@
|
|||||||
|
/*
|
||||||
|
* 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.relay.filters.Filter
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||||
|
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
||||||
|
import kotlin.test.Test
|
||||||
|
import kotlin.test.assertEquals
|
||||||
|
import kotlin.test.assertTrue
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Verifies the NIP-77 negentropy id-and-time projection against the
|
||||||
|
* full-event query path. Goal: same result set, ~25× lighter
|
||||||
|
* footprint per row. Run across every indexing strategy via
|
||||||
|
* [BaseDBTest.forEachDB] so plan changes don't silently break the
|
||||||
|
* snapshot path.
|
||||||
|
*/
|
||||||
|
class SnapshotIdsForNegentropyTest : BaseDBTest() {
|
||||||
|
private val signer = NostrSignerSync()
|
||||||
|
|
||||||
|
private fun makeEvents(count: Int) =
|
||||||
|
List(count) { i ->
|
||||||
|
signer.sign(TextNoteEvent.build("event-$i", createdAt = 1_700_000_000L + i))
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun matchesFullQueryForSimpleKindFilter() =
|
||||||
|
forEachDB { db ->
|
||||||
|
val events = makeEvents(50)
|
||||||
|
for (e in events) db.insert(e)
|
||||||
|
|
||||||
|
val filter = Filter(kinds = listOf(1))
|
||||||
|
val full = db.query<com.vitorpamplona.quartz.nip01Core.core.Event>(filter)
|
||||||
|
val ids = db.snapshotIdsForNegentropy(listOf(filter))
|
||||||
|
|
||||||
|
assertEquals(full.size, ids.size, "snapshot must cover the same row set")
|
||||||
|
assertEquals(
|
||||||
|
full.map { it.id }.toSet(),
|
||||||
|
ids.map { it.id }.toSet(),
|
||||||
|
"snapshot ids must match the full-query ids",
|
||||||
|
)
|
||||||
|
// Every (createdAt, id) pair must round-trip.
|
||||||
|
val byId = full.associate { it.id to it.createdAt }
|
||||||
|
for (entry in ids) {
|
||||||
|
assertEquals(byId[entry.id], entry.createdAt, "createdAt mismatch for ${entry.id}")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun honorsSinceUntilLimit() =
|
||||||
|
forEachDB { db ->
|
||||||
|
val events = makeEvents(20) // createdAt 1_700_000_000..1_700_000_019
|
||||||
|
for (e in events) db.insert(e)
|
||||||
|
|
||||||
|
// since/until window: [+5, +14] inclusive
|
||||||
|
val filter =
|
||||||
|
Filter(
|
||||||
|
kinds = listOf(1),
|
||||||
|
since = 1_700_000_005L,
|
||||||
|
until = 1_700_000_014L,
|
||||||
|
)
|
||||||
|
val ids = db.snapshotIdsForNegentropy(listOf(filter))
|
||||||
|
assertEquals(10, ids.size, "since/until window should yield 10 rows")
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun maxEntriesPlusOneSentinelMarksOverflow() =
|
||||||
|
forEachDB { db ->
|
||||||
|
val events = makeEvents(30)
|
||||||
|
for (e in events) db.insert(e)
|
||||||
|
|
||||||
|
val filter = Filter(kinds = listOf(1))
|
||||||
|
// cap = 10; we have 30 rows, so the result must be 11
|
||||||
|
// (cap + 1 sentinel) — matches strfry's `maxSyncEvents`
|
||||||
|
// overflow-detection idiom.
|
||||||
|
val capped = db.snapshotIdsForNegentropy(listOf(filter), maxEntries = 10)
|
||||||
|
assertEquals(11, capped.size)
|
||||||
|
assertTrue(capped.size > 10, "caller relies on size > cap as overflow signal")
|
||||||
|
|
||||||
|
// cap >= total: returns the whole set unchanged.
|
||||||
|
val whole = db.snapshotIdsForNegentropy(listOf(filter), maxEntries = 100)
|
||||||
|
assertEquals(30, whole.size)
|
||||||
|
}
|
||||||
|
}
|
||||||
+1
-1
@@ -285,7 +285,7 @@ class NegentropySessionTest {
|
|||||||
val openCmd = clientSession.open()
|
val openCmd = clientSession.open()
|
||||||
|
|
||||||
// Server processes via NegentropyServerSession
|
// Server processes via NegentropyServerSession
|
||||||
val serverSession = NegentropyServerSession("sub1", serverEvents)
|
val serverSession = NegentropyServerSession.fromEvents("sub1", serverEvents)
|
||||||
val response = serverSession.processMessage(openCmd.initialMessage)
|
val response = serverSession.processMessage(openCmd.initialMessage)
|
||||||
|
|
||||||
assertNotNull(response)
|
assertNotNull(response)
|
||||||
|
|||||||
Reference in New Issue
Block a user