mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-10 00:16:59 +00:00
Merge pull request #2771 from vitorpamplona/claude/negentropy-strfry-interop-4y5hm
NIP-77: strfry-interop snapshot path with bounded sync
This commit is contained in:
Vitor Pamplona
@@ -1,20 +1,20 @@
|
||||
# NIP-77 negentropy at scale: snapshot memory + chunked replay
|
||||
# NIP-77 negentropy at scale: strfry-interop snapshot path
|
||||
|
||||
## Problem
|
||||
|
||||
`RelaySession` delegates NEG-OPEN to `NegSessionRegistry.open`
|
||||
(`quartz/nip01Core/relay/server/NegSessionRegistry.kt`), which calls
|
||||
`store.snapshotQuery(filters)` and feeds the **entire** result list
|
||||
into `NegentropyServerSession`. For a relay holding 5 M events that
|
||||
match a broad NEG-OPEN filter (`{kinds: [1, 7]}`), this is 5 M
|
||||
`Event` objects materialised in memory before the first NEG-MSG goes
|
||||
out.
|
||||
(`quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/NegSessionRegistry.kt:58-77`),
|
||||
which calls `store.snapshotQuery(filters)` and feeds the **entire**
|
||||
result list of full `Event` objects into `NegentropyServerSession`
|
||||
(`quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip77Negentropy/NegentropyServerSession.kt:40-54`).
|
||||
For a relay holding 5 M events that match a broad NEG-OPEN filter
|
||||
(`{kinds: [1, 7]}`), this materialises 5 M `Event` objects with
|
||||
content/tags/sig — call it ~1 KB/event, ~5 GB transient pressure per
|
||||
concurrent NEG-OPEN — before the first NEG-MSG goes out.
|
||||
|
||||
The negentropy library itself is fine — it pivots into a sealed
|
||||
`StorageVector` (id + createdAt only, ~40 bytes/entry). But the
|
||||
`store.query<Event>(f)` step that produces the input materialises full
|
||||
`Event` objects with content, tags, sig — call it ~1 KB/event. 5 M ×
|
||||
1 KB = 5 GB transient pressure per concurrent NEG-OPEN.
|
||||
The negentropy library itself is fine. Internally it pivots into a
|
||||
sealed `StorageVector` of `(uint64 timestamp, byte[32] id)` items —
|
||||
40 bytes/entry. The waste is purely in the snapshot step.
|
||||
|
||||
Two operator-visible symptoms:
|
||||
|
||||
@@ -24,77 +24,221 @@ Two operator-visible symptoms:
|
||||
large stores the client waits seconds for what should be a
|
||||
millisecond round-trip.
|
||||
|
||||
## Reference: strfry
|
||||
|
||||
We want byte-for-byte interop and comparable throughput against
|
||||
[hoytech/strfry](https://github.com/hoytech/strfry). The relevant
|
||||
defaults from `strfry/src/apps/relay/RelayNegentropy.cpp` and
|
||||
`golpe.yaml`:
|
||||
|
||||
| Knob | strfry default | Notes |
|
||||
|---------------------------------------|-----------------------|-------|
|
||||
| `frameSizeLimit` (NEG-MSG payload) | **500_000 bytes** | Hard-coded `Negentropy ne(storage, 500'000)` |
|
||||
| `relay__negentropy__maxSyncEvents` | **1_000_000** | Hard cap on items in the snapshot |
|
||||
| `relay__maxSubsPerConnection` | **200** | Shared between REQ and NEG sessions |
|
||||
| `idSize` on the wire | **32 bytes** | NIP-77 v1 (`PROTOCOL_VERSION = 0x61`) |
|
||||
| Default `since` window | **none** | Filter is honored as-is |
|
||||
| Filter parser | same as REQ | Honors `limit`, `kinds`, `authors`, `#tags` |
|
||||
| Snapshot data | `(created_at, id)` only | LMDB scan inserts into `negentropy::storage::Vector` |
|
||||
|
||||
Two-phase materialisation in strfry: `QueryScheduler` scans LMDB
|
||||
asynchronously, batching matched level-ids into a `vector<uint64_t>`;
|
||||
on completion the worker pulls each event header, calls
|
||||
`storageVector.insert(packed.created_at(), packed.id())`, then
|
||||
`seal()`s. The session response is sent only after seal.
|
||||
|
||||
Our equivalent must (a) match the snapshot footprint of ~40 bytes per
|
||||
event, and (b) match the wire-level frame-cap so a single
|
||||
reconciliation round-trip exchanges the same payload size.
|
||||
|
||||
## Sketch
|
||||
|
||||
### A — id-and-time-only snapshot path
|
||||
### A — id-and-time-only snapshot path (memory parity)
|
||||
|
||||
Negentropy only needs `(createdAt, id)` pairs. Add a streaming
|
||||
`IEventStore.queryIdAndTime(filter)` that returns
|
||||
`Sequence<Pair<Long, ByteArray>>` (or a `Flow` of small chunks) —
|
||||
no content/tags/sig, no Event allocation. SQLite path is a SELECT
|
||||
on `event_headers` (the `created_at`, `id` columns are already
|
||||
indexed for query plans).
|
||||
`IEventStore.snapshotIdsForNegentropy(filter)` that returns
|
||||
`Sequence<IdAndTime>` (`data class IdAndTime(val createdAt: Long, val id: ByteArray)`)
|
||||
— no content/tags/sig, no `Event` allocation. The SQLite path is a
|
||||
plain `SELECT id, created_at FROM event_headers WHERE …` against the
|
||||
existing `query_by_created_at_id` index
|
||||
(`quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventIndexesModule.kt:79`).
|
||||
|
||||
```kotlin
|
||||
suspend fun snapshotIdsForNegentropy(filter: Filter): IdTimeStream
|
||||
// IEventStore
|
||||
suspend fun snapshotIdsForNegentropy(filters: List<Filter>): List<IdAndTime>
|
||||
|
||||
// LiveEventStore — already deduplicates union of multi-filter results
|
||||
suspend fun snapshotIdsForNegentropy(filters: List<Filter>): List<IdAndTime>
|
||||
```
|
||||
|
||||
`NegentropyServerSession` is rewritten to take that stream and feed
|
||||
it directly into the `StorageVector`. Memory drops from O(N × 1 KB)
|
||||
to O(N × 40 B) — a 25× reduction; for 5 M events, ~200 MB instead
|
||||
of 5 GB.
|
||||
`NegentropyServerSession` is rewritten to take that list (or a
|
||||
two-phase `pendingIds → seal` builder) and feed it directly into
|
||||
`StorageVector`. Memory drops from O(N × ~1 KB) to O(N × 40 B) — for
|
||||
1 M events, ~40 MB instead of ~1 GB; matches strfry's per-session
|
||||
footprint.
|
||||
|
||||
### B — bounded-window subscriptions
|
||||
**id encoding:** strfry stores 32-byte raw ids (NIP-77 v1
|
||||
`ID_SIZE = 32`); the 16-byte truncation is `FINGERPRINT_SIZE`, only used
|
||||
internally for SHA-256 accumulator output. The current Kotlin code
|
||||
already passes the hex `event.id` string to `storage.insert(..)`
|
||||
(`NegentropyServerSession.kt:50`); the kmp-negentropy library decodes
|
||||
that to 32 raw bytes. Confirm this is preserved when we switch to a
|
||||
`ByteArray` id input — do not pre-truncate.
|
||||
|
||||
Most NEG-OPENs from real Nostr clients want the last 30 days, not
|
||||
"everything." If the client doesn't supply `since`, the server can
|
||||
default to a configurable horizon (e.g. 90 days) and surface this in
|
||||
the NIP-11 `limitation.negentropy_max_lookback_seconds` field.
|
||||
Operators can lift the cap; clients reading the doc know the bound.
|
||||
### B — bounded snapshot size, NOT a default `since` window
|
||||
|
||||
This is a NIP-spec-adjacent question more than a code change — needs
|
||||
a comment on whether the spec allows it. nostr-rs-relay does this
|
||||
already.
|
||||
The previous draft of this plan suggested defaulting `since` to a 90-day
|
||||
horizon. **Drop that.** strfry doesn't do it — the filter is honored
|
||||
as-is — and silently bounding `since` would break interop with strfry's
|
||||
sync clients (e.g. `strfry sync`, nostr-sdk's negentropy reconciler):
|
||||
they ask for "everything" and rely on getting it.
|
||||
|
||||
### C — frame-size cap on NEG-MSG
|
||||
Instead match strfry's protection: a hard cap on the number of items
|
||||
that go into a single snapshot.
|
||||
|
||||
```toml
|
||||
[negentropy]
|
||||
max_sync_events = 1_000_000 # matches strfry's relay__negentropy__maxSyncEvents
|
||||
```
|
||||
|
||||
`NegSessionRegistry.open` checks `count >= max_sync_events` after the
|
||||
SQLite count or during scan, and on overflow sends:
|
||||
|
||||
```
|
||||
["NEG-ERR", "<subId>", "blocked: too many query results"]
|
||||
```
|
||||
|
||||
Exact wording matches strfry so client error-handling that string-matches
|
||||
behaves identically. (Spec doesn't normatively define error text; this is
|
||||
de-facto interop.)
|
||||
|
||||
### C — frame-size cap on NEG-MSG: 500_000 bytes (strfry parity)
|
||||
|
||||
`NegentropyServerSession` is constructed with `frameSizeLimit = 0`
|
||||
(no limit). At very large reconciliations the message can grow large.
|
||||
Set a default `frameSizeLimit = 64 * 1024` (matching the typical WS
|
||||
frame budget) so NEG-MSGs don't blow past `[limits].max_ws_frame_bytes`.
|
||||
today. **Set `frameSizeLimit = 500_000`** (matching strfry) so a single
|
||||
NEG-MSG round-trip carries the same payload as strfry's. 64 KB — what
|
||||
the previous draft suggested — would force 8× more round-trips for
|
||||
large reconciliations and noticeably slow sync against strfry-style
|
||||
clients that expect ~1 MB hex-encoded NEG-MSGs.
|
||||
|
||||
The library already supports this — pure config change in
|
||||
`NegSessionRegistry.open`.
|
||||
The kmp-negentropy library enforces `frameSizeLimit >= 4096` (or `0`
|
||||
for unlimited). 500_000 is well above the floor. Pure config change in
|
||||
`NegSessionRegistry.open`; expose via `[negentropy].frame_size_limit`
|
||||
for operators who tune it down to fit smaller WS frame budgets.
|
||||
|
||||
### D — concurrent NEG-OPEN cap
|
||||
Note: `LimitsSection.max_ws_frame_bytes`
|
||||
(`geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt:148`)
|
||||
applies to the WebSocket layer. After hex-encoding a 500_000-byte
|
||||
negentropy payload doubles to ~1_000_000 bytes on the wire; ensure
|
||||
`max_ws_frame_bytes` is at least 2 MB (or unlimited) in default
|
||||
config so the response isn't truncated by the WS layer.
|
||||
|
||||
### D — concurrent NEG-OPEN cap, shared with REQ
|
||||
|
||||
A NEG-OPEN holds session state until NEG-CLOSE (or connection close).
|
||||
Today nothing caps the number of concurrent open negentropy sessions
|
||||
per connection. A misbehaving (or hostile) client could open thousands
|
||||
and pin RAM. Add `MAX_NEG_SESSIONS_PER_CONNECTION = 16`, send NEG-ERR
|
||||
on overflow.
|
||||
Today `NegSessionRegistry.sessions` is unbounded. strfry caps at 200
|
||||
sessions per connection **shared with REQ** — both count against
|
||||
`relay__maxSubsPerConnection`.
|
||||
|
||||
Implement as:
|
||||
- Reuse the existing per-connection REQ subscription cap (or introduce
|
||||
one if absent) and let NEG sessions consume the same budget.
|
||||
- Default cap: **200** to match strfry. Configurable via
|
||||
`[limits].max_subs_per_connection`.
|
||||
- On overflow strfry sends a NOTICE, not NEG-ERR:
|
||||
`["NOTICE", "too many concurrent NEG requests"]`. We should match
|
||||
this — `NEG-ERR` is reserved for per-session protocol errors in
|
||||
strfry's model.
|
||||
|
||||
### E — pre-built fingerprint tree (follow-up, not in scope here)
|
||||
|
||||
strfry's real production-scale advantage is the **`negentropy::storage::BTreeLMDB`** backend: a persistent, incrementally-maintained B-tree
|
||||
of `(timestamp, id)` keys with per-node fingerprint accumulators.
|
||||
When NEG-OPEN's filter string matches a pre-registered
|
||||
`NegentropyFilter` tree, strfry skips the materialise-and-seal step
|
||||
entirely and reconciles directly off the LMDB B-tree in O(log n)
|
||||
fingerprint computations. See `env.foreach_NegentropyFilter` and
|
||||
`addStatelessView` in `RelayNegentropy.cpp`.
|
||||
|
||||
Equivalent for geode: a `quartz/.../nip77Negentropy/PrebuiltStorage`
|
||||
backed by a SQLite-side incremental fingerprint index, registered per
|
||||
canonical filter. This is a substantial piece of work and **out of
|
||||
scope for this plan** — call it out for a follow-up
|
||||
(`geode/plans/2026-05-08-negentropy-prebuilt-tree.md`). The A+B+C+D
|
||||
combination is enough to match strfry's `MemoryView` path, which is
|
||||
what 99% of ad-hoc NEG-OPENs hit.
|
||||
|
||||
## Concrete error-string interop table
|
||||
|
||||
Match strfry verbatim — clients in the wild string-match on these:
|
||||
|
||||
| Condition | strfry frame |
|
||||
|---------------------------------------|-----------------------------------------------------------|
|
||||
| Snapshot exceeds `max_sync_events` | `["NEG-ERR", "<subId>", "blocked: too many query results"]` |
|
||||
| NEG-MSG for unknown subId | `["NEG-ERR", "<subId>", "closed: unknown subscription handle"]` |
|
||||
| Library `reconcile()` parse failure | `["NEG-ERR", "<subId>", "PROTOCOL-ERROR"]` |
|
||||
| Per-connection sub cap exceeded | `["NOTICE", "too many concurrent NEG requests"]` |
|
||||
| NEG-MSG before NEG-OPEN seal complete | `["NOTICE", "negentropy error: got NEG-MSG before NEG-OPEN complete"]` |
|
||||
|
||||
The current Kotlin code in `NegSessionRegistry.kt:82` sends
|
||||
`"error: no negentropy session for <subId>"` for the unknown-subId
|
||||
case. Update to strfry's wording.
|
||||
|
||||
## Wire-level conformance checks
|
||||
|
||||
Before merging, verify against strfry as ground truth:
|
||||
|
||||
1. **Round-trip with `strfry sync`**: stand up a small geode instance,
|
||||
point `strfry sync ws://geode-host` at it, confirm sync completes
|
||||
and converges in ≤ comparable round-trips for an N=100k corpus.
|
||||
2. **idSize**: assert kmp-negentropy round-trips full 32-byte ids;
|
||||
the 16-byte fingerprint stays internal to the library.
|
||||
3. **Protocol version byte**: NIP-77 v1 = `0x61`. Confirm
|
||||
`NegentropyServerSession.processMessage` neither emits nor accepts
|
||||
a different version byte. Negotiation happens inside the library;
|
||||
surface the version on `NIP-11.limitation.negentropy = 1` so clients
|
||||
know the v1 path is supported.
|
||||
|
||||
## How to verify
|
||||
|
||||
Add to `geode.perf.LoadBenchmark`:
|
||||
Add to `geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt`:
|
||||
|
||||
- `negentropyOpenLatencyLargeCorpus` — preload 1 M events (use
|
||||
fixtures), measure NEG-OPEN → first NEG-MSG latency. Target <100 ms.
|
||||
- `negentropyOpenLatencyLargeCorpus` — preload 1 M events, measure
|
||||
NEG-OPEN → first NEG-MSG latency. Target **<200 ms** (strfry's
|
||||
C++ `MemoryView` path on equivalent hardware does ~80–150 ms;
|
||||
we expect a 1.5–2× JVM tax).
|
||||
- `negentropyMemoryPressure` — open 10 concurrent NEG-OPENs on the
|
||||
same large corpus; measure RSS delta, target <500 MB.
|
||||
same large corpus; measure RSS delta. Target **<500 MB**
|
||||
(10 × ~40 MB session footprint + scan overhead).
|
||||
- `negentropyStrfryInterop` — programmatic round-trip against a
|
||||
containerised `hoytech/strfry`: same fixture corpus loaded in both,
|
||||
cross-sync, assert byte-identical id-set convergence in equal
|
||||
round-trips ±1.
|
||||
|
||||
Add to `quartz/.../nip77Negentropy/`:
|
||||
|
||||
- `NegentropyServerSessionTest.processMessage_atFrameLimit_splitsAcrossRounds`
|
||||
— large symmetric difference, assert each NEG-MSG payload is
|
||||
≤ 500_000 bytes and the session completes in N rounds (not 1).
|
||||
|
||||
## Risks
|
||||
|
||||
- **`Sequence`/`Flow` over SQLite cursor**: holding a cursor open
|
||||
across the full sync is fragile if the client stalls. Materialise
|
||||
to a smaller in-memory list (just (id, createdAt)) once, reuse for
|
||||
the lifetime of the session. Memory bound is the same.
|
||||
- **Defaulting `since` is a behaviour change**: existing clients that
|
||||
expect "everything" silently get a bounded window. Either (a) make
|
||||
it opt-in via `RelayConfig.NegentropySection.default_lookback_seconds
|
||||
= null`, (b) advertise the cap in NIP-11 so well-behaved clients
|
||||
read it.
|
||||
- **Frame-size cap can break older clients**: the NIP-77 reference
|
||||
implementation (kmp-negentropy) handles this gracefully — multi-frame
|
||||
reconciliation is in spec — but field-test against a known-working
|
||||
client (e.g. nstart, primal-cache) before flipping the default.
|
||||
- **Cursor lifetime**: holding a SQLite cursor open across the full
|
||||
sync is fragile if the client stalls. Materialise to a smaller
|
||||
in-memory `List<IdAndTime>` (40 bytes/entry) once at NEG-OPEN time,
|
||||
reuse for the lifetime of the session. Bounded by `max_sync_events`.
|
||||
- **Frame-size 500_000 vs WS frame budget**: hex-encoded payload is
|
||||
~1 MB. If `LimitsSection.max_ws_frame_bytes` is set lower than
|
||||
~1.5 MB the response gets truncated/rejected by the WS layer. Lift
|
||||
the WS default OR cap `frame_size_limit` to
|
||||
`max_ws_frame_bytes / 2` at startup; fail-fast log a warning if the
|
||||
operator's config makes negentropy unusable.
|
||||
- **No default `since` (intentional, but worth flagging)**: a hostile
|
||||
client doing `NEG-OPEN {kinds:[1]}` against a large corpus will hit
|
||||
`max_sync_events` and get NEG-ERR. The cap is the protection; do
|
||||
not also silently bound `since`.
|
||||
- **kmp-negentropy library version**: pinned to `v1.0.2`
|
||||
(`gradle/libs.versions.toml:9`). Confirm v1.0.2 enforces
|
||||
`frameSizeLimit >= 4096`, supports protocol byte `0x61`, and
|
||||
internal `ID_SIZE = 32`. If any of those don't match strfry, this
|
||||
plan needs an upstream fix on kmp-negentropy first.
|
||||
|
||||
@@ -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.store.IEventStore
|
||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||
import java.io.File
|
||||
|
||||
/**
|
||||
@@ -109,6 +110,12 @@ fun main(args: Array<String>) {
|
||||
}
|
||||
|
||||
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 =
|
||||
Relay(
|
||||
advertisedUrl,
|
||||
@@ -117,6 +124,7 @@ fun main(args: Array<String>) {
|
||||
policyBuilder,
|
||||
stateFile = stateFile,
|
||||
parallelVerify = parallelVerify,
|
||||
negentropySettings = negentropySettings,
|
||||
)
|
||||
// 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
|
||||
|
||||
@@ -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.sqlite.EventStore
|
||||
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.BanStore
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
@@ -84,6 +85,11 @@ class Relay(
|
||||
* `Main.kt` skips `VerifyPolicy` when this flag is on.
|
||||
*/
|
||||
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 {
|
||||
private val stateStore: RelayStateStore? = stateFile?.let { RelayStateStore(it) }
|
||||
|
||||
@@ -168,8 +174,9 @@ class Relay(
|
||||
val user = policyBuilder()
|
||||
if (user === EmptyPolicy) BanListPolicy(banStore) else user + BanListPolicy(banStore)
|
||||
},
|
||||
parentContext,
|
||||
parentContext = parentContext,
|
||||
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.WebSocketListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
@@ -50,6 +51,7 @@ import java.util.concurrent.ConcurrentHashMap
|
||||
*/
|
||||
class RelayHub(
|
||||
private val defaultPolicy: () -> IRelayPolicy = { EmptyPolicy },
|
||||
private val negentropySettings: NegentropySettings = NegentropySettings.Default,
|
||||
) : WebsocketBuilder,
|
||||
AutoCloseable {
|
||||
private val relays = ConcurrentHashMap<NormalizedRelayUrl, Relay>()
|
||||
@@ -60,7 +62,11 @@ class RelayHub(
|
||||
fun getOrCreate(url: NormalizedRelayUrl): Relay {
|
||||
check(!closed) { "RelayHub has been closed" }
|
||||
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 authorization: AuthorizationSection = AuthorizationSection(),
|
||||
val admin: AdminSection = AdminSection(),
|
||||
val negentropy: NegentropySection = NegentropySection(),
|
||||
) {
|
||||
/**
|
||||
* 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,
|
||||
)
|
||||
|
||||
/**
|
||||
* 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(
|
||||
val pubkey_whitelist: 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.crypto.KeyPair
|
||||
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.normalizer.NormalizedRelayUrl
|
||||
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.NegMsgMessage
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
|
||||
import kotlinx.coroutines.runBlocking
|
||||
@@ -237,12 +239,82 @@ class Nip77NegentropyTest {
|
||||
val response = client.nextMessage()
|
||||
assertTrue(response is NegErrMessage, "expected NEG-ERR, got ${response::class.simpleName}")
|
||||
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 {
|
||||
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
|
||||
fun negOpenWithSameSubIdReplacesPriorSession() =
|
||||
runBlocking {
|
||||
|
||||
+186
@@ -0,0 +1,186 @@
|
||||
/*
|
||||
* 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.geode.interop
|
||||
|
||||
import com.vitorpamplona.geode.LocalRelayServer
|
||||
import com.vitorpamplona.geode.Relay
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import okhttp3.OkHttpClient
|
||||
import kotlin.test.AfterTest
|
||||
import kotlin.test.BeforeTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* The Kotlin counterpart to strfry's `test/syncTest.pl`: stand up two
|
||||
* Geode relays on real WebSocket endpoints, give each a partially
|
||||
* overlapping corpus, and converge them via NIP-77.
|
||||
*
|
||||
* The driver mirrors `strfry sync ws://other --dir both`:
|
||||
*
|
||||
* 1. Read the local relay's snapshot for the negotiated filter.
|
||||
* 2. Open NEG-OPEN against the remote with that snapshot.
|
||||
* 3. Drive NEG-MSG round trips until the client-side
|
||||
* [com.vitorpamplona.quartz.nip77Negentropy.NegentropySession]
|
||||
* reports completion.
|
||||
* 4. `needIds`: REQ them from the remote, insert into the local relay.
|
||||
* 5. `haveIds`: fetch from the local relay, publish to the remote.
|
||||
*
|
||||
* After the round, both relays must hold the union of the original
|
||||
* corpora. We assert via REQ on each side; an `id`-filter that returns
|
||||
* every event we expect, and nothing more, proves convergence
|
||||
* end-to-end through the NIP-77 server pipeline (`NegSessionRegistry`
|
||||
* → `NegentropyServerSession` → `IEventStore.snapshotIdsForNegentropy`).
|
||||
*
|
||||
* Equivalent to strfry's `runSyncTests.pl` "full DB sync" case at
|
||||
* small scale. Larger corpora belong in `LoadBenchmark`.
|
||||
*/
|
||||
class GeodeVsGeodeNegentropySyncTest {
|
||||
private lateinit var relayA: Relay
|
||||
private lateinit var relayB: Relay
|
||||
private lateinit var serverA: LocalRelayServer
|
||||
private lateinit var serverB: LocalRelayServer
|
||||
private lateinit var scope: CoroutineScope
|
||||
private lateinit var client: NostrClient
|
||||
private val httpClient = OkHttpClient.Builder().build()
|
||||
|
||||
@BeforeTest
|
||||
fun setup() {
|
||||
// The placeholder URLs are normalised so the relay accepts
|
||||
// them; ports come from the autobind below via [server.url].
|
||||
relayA = Relay(url = "ws://127.0.0.1:7771/".normalizeRelayUrl())
|
||||
relayB = Relay(url = "ws://127.0.0.1:7772/".normalizeRelayUrl())
|
||||
serverA = LocalRelayServer(relayA, host = "127.0.0.1", port = 0).start()
|
||||
serverB = LocalRelayServer(relayB, host = "127.0.0.1", port = 0).start()
|
||||
scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
|
||||
client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> httpClient }, scope)
|
||||
}
|
||||
|
||||
@AfterTest
|
||||
fun teardown() {
|
||||
client.disconnect()
|
||||
scope.cancel()
|
||||
serverA.stop()
|
||||
serverB.stop()
|
||||
relayA.close()
|
||||
relayB.close()
|
||||
}
|
||||
|
||||
/** Generates [count] signed text notes with monotonic createdAt. */
|
||||
private fun makeEvents(
|
||||
count: Int,
|
||||
seed: Long = 1_700_000_000L,
|
||||
): List<Event> {
|
||||
val signer = NostrSignerSync(KeyPair())
|
||||
return List(count) { i ->
|
||||
signer.sign(TextNoteEvent.build("event-$seed-$i", createdAt = seed + i))
|
||||
}
|
||||
}
|
||||
|
||||
private val driver by lazy { InteropSyncDriver(httpClient) }
|
||||
|
||||
@Test
|
||||
fun bidirectionalSyncConvergesTwoRelays() =
|
||||
runBlocking {
|
||||
// Universe of 20 events. Relay A holds [0..14], Relay B
|
||||
// holds [5..19]. Overlap [5..14], A-only [0..4], B-only
|
||||
// [15..19]. After bidirectional sync both must hold [0..19].
|
||||
val all = makeEvents(20)
|
||||
val aEvents = all.subList(0, 15)
|
||||
val bEvents = all.subList(5, 20)
|
||||
relayA.preload(aEvents)
|
||||
relayB.preload(bEvents)
|
||||
|
||||
val filter = Filter(kinds = listOf(1))
|
||||
val urlA = serverA.url.normalizeRelayUrl()
|
||||
val urlB = serverB.url.normalizeRelayUrl()
|
||||
|
||||
// B negotiates the symmetric difference with A.
|
||||
val diff = driver.negotiate(serverA.url, filter, bEvents)
|
||||
assertNull(diff.error, "negotiation must not error: ${diff.error}")
|
||||
// From B's perspective: needs A-only, has B-only.
|
||||
assertEquals(
|
||||
aEvents.subList(0, 5).map { it.id }.toSet(),
|
||||
diff.needIds,
|
||||
"B should NEED [0..4] from A",
|
||||
)
|
||||
assertEquals(
|
||||
bEvents.subList(10, 15).map { it.id }.toSet(),
|
||||
diff.haveIds,
|
||||
"B should announce HAVE for [15..19]",
|
||||
)
|
||||
|
||||
// Close the loop: fetch needs from A, push haves to A.
|
||||
val needFromA =
|
||||
client.fetchAll(relay = urlA, filter = Filter(ids = diff.needIds.toList()))
|
||||
relayB.preload(needFromA)
|
||||
|
||||
for (id in diff.haveIds) {
|
||||
client.publishAndConfirm(bEvents.first { it.id == id }, setOf(urlA))
|
||||
}
|
||||
|
||||
// --- Verify convergence ---
|
||||
val expected = all.map { it.id }.toSet()
|
||||
assertEquals(expected, idsOnRelay(urlA, filter), "Relay A should hold every event")
|
||||
assertEquals(expected, idsOnRelay(urlB, filter), "Relay B should hold every event")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun negentropyConvergesInBoundedRoundsOnSmallCorpus() =
|
||||
runBlocking {
|
||||
val all = makeEvents(200)
|
||||
relayA.preload(all.subList(0, 150))
|
||||
val bEvents = all.subList(50, 200)
|
||||
relayB.preload(bEvents)
|
||||
|
||||
val res = driver.negotiate(serverA.url, Filter(kinds = listOf(1)), bEvents)
|
||||
assertNull(res.error)
|
||||
assertEquals(50, res.needIds.size, "B should need [0..49]")
|
||||
assertEquals(50, res.haveIds.size, "B should have [150..199]")
|
||||
// strfry typically converges these in ≤5 rounds; ≤16 is
|
||||
// generous headroom that still catches regressions.
|
||||
assertTrue(res.rounds <= 16, "expected ≤16 NEG-MSG rounds, got ${res.rounds}")
|
||||
}
|
||||
|
||||
/** Every event id matching [filter] visible on the relay at [url] via REQ. */
|
||||
private suspend fun idsOnRelay(
|
||||
url: NormalizedRelayUrl,
|
||||
filter: Filter,
|
||||
): Set<HexKey> = client.fetchAll(relay = url, filter = filter).map { it.id }.toSet()
|
||||
}
|
||||
+220
@@ -0,0 +1,220 @@
|
||||
/*
|
||||
* 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.geode.interop
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
|
||||
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
import okhttp3.Response
|
||||
import okhttp3.WebSocket
|
||||
import okhttp3.WebSocketListener
|
||||
import java.io.File
|
||||
import java.net.ServerSocket
|
||||
import java.net.Socket
|
||||
import kotlin.io.path.createTempDirectory
|
||||
import kotlin.test.AfterTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* Reciprocal interop test: Geode's NIP-77 client driving a real
|
||||
* `strfry` instance.
|
||||
*
|
||||
* **Opt-in.** Skipped unless `STRFRY_BIN` env var (or
|
||||
* `-Dstrfry.bin=...`) points at a `strfry` binary. When unset, the
|
||||
* test prints a `[skip]` line and returns. Mirrors the gate
|
||||
* `LoadBenchmark` uses for `-DrunLoadBenchmark`:
|
||||
*
|
||||
* STRFRY_BIN=/usr/local/bin/strfry ./gradlew :geode:test \
|
||||
* --tests "*GeodeVsStrfryNegentropySyncTest*"
|
||||
*
|
||||
* The strfry process is booted on a free loopback port with a fresh
|
||||
* temp LMDB dir. We feed events into it via the NIP-01 EVENT wire
|
||||
* (no `strfry import`), then run [InteropSyncDriver] against it.
|
||||
* Strfry's `RelayNegentropy.cpp` answers the same NIP-77 wire we test
|
||||
* against Geode — passing both is byte-shape interop, not just
|
||||
* "passes our own tests".
|
||||
*/
|
||||
class GeodeVsStrfryNegentropySyncTest {
|
||||
private val strfryBin: String? =
|
||||
System.getenv("STRFRY_BIN") ?: System.getProperty("strfry.bin")
|
||||
private val enabled = strfryBin != null
|
||||
|
||||
private lateinit var strfryDir: File
|
||||
private var strfryProcess: Process? = null
|
||||
private val httpClient by lazy { OkHttpClient.Builder().build() }
|
||||
|
||||
@AfterTest
|
||||
fun teardown() {
|
||||
strfryProcess?.destroy()
|
||||
strfryProcess?.waitFor()
|
||||
if (::strfryDir.isInitialized) strfryDir.deleteRecursively()
|
||||
}
|
||||
|
||||
/**
|
||||
* Boots a strfry instance in a temp LMDB dir on a free port.
|
||||
* Writes the smallest config strfry accepts, starts the
|
||||
* subprocess, and polls until the WebSocket port is reachable.
|
||||
*
|
||||
* The config is intentionally minimal — strfry's defaults
|
||||
* (negentropy on, sane limits) are what we want to test against.
|
||||
* Adding speculative config keys risks failing on schema drift.
|
||||
*/
|
||||
private fun startStrfry(): String {
|
||||
val port = ServerSocket(0).use { it.localPort }
|
||||
strfryDir = createTempDirectory(prefix = "strfry-interop-").toFile()
|
||||
val configFile = File(strfryDir, "strfry.conf")
|
||||
configFile.writeText(
|
||||
"""
|
||||
db = "${strfryDir.absolutePath}/strfry-db"
|
||||
relay {
|
||||
bind = "127.0.0.1"
|
||||
port = $port
|
||||
}
|
||||
""".trimIndent(),
|
||||
)
|
||||
File(strfryDir, "strfry-db").mkdirs()
|
||||
strfryProcess =
|
||||
ProcessBuilder(strfryBin, "--config", configFile.absolutePath, "relay")
|
||||
.redirectErrorStream(true)
|
||||
.redirectOutput(File(strfryDir, "strfry.log"))
|
||||
.start()
|
||||
|
||||
val deadline = System.currentTimeMillis() + 5_000
|
||||
while (System.currentTimeMillis() < deadline) {
|
||||
runCatching {
|
||||
Socket("127.0.0.1", port).close()
|
||||
return "ws://127.0.0.1:$port/"
|
||||
}
|
||||
Thread.sleep(100)
|
||||
}
|
||||
throw IllegalStateException(
|
||||
"strfry did not start within 5s; log: " +
|
||||
File(strfryDir, "strfry.log").readText(),
|
||||
)
|
||||
}
|
||||
|
||||
private fun makeEvents(count: Int): List<Event> {
|
||||
val signer = NostrSignerSync(KeyPair())
|
||||
val now = 1_700_000_000L
|
||||
return List(count) { i ->
|
||||
signer.sign(TextNoteEvent.build("strfry-interop-$i", createdAt = now + i))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Push every event in [events] to the relay at [wsUrl] over a
|
||||
* one-shot WebSocket, waiting for an `OK` response per event.
|
||||
*
|
||||
* Used to seed strfry from the same `Event` objects Geode's
|
||||
* `Relay.preload` accepts — that way both sides start from a
|
||||
* byte-identical corpus.
|
||||
*/
|
||||
private suspend fun publishToStrfry(
|
||||
wsUrl: String,
|
||||
events: List<Event>,
|
||||
) {
|
||||
val incoming = Channel<String>(UNLIMITED)
|
||||
val ws =
|
||||
httpClient.newWebSocket(
|
||||
Request.Builder().url(wsUrl.replace("ws://", "http://")).build(),
|
||||
object : WebSocketListener() {
|
||||
override fun onMessage(
|
||||
webSocket: WebSocket,
|
||||
text: String,
|
||||
) {
|
||||
incoming.trySend(text)
|
||||
}
|
||||
|
||||
override fun onFailure(
|
||||
webSocket: WebSocket,
|
||||
t: Throwable,
|
||||
response: Response?,
|
||||
) {
|
||||
incoming.close(t)
|
||||
}
|
||||
},
|
||||
)
|
||||
try {
|
||||
for (e in events) {
|
||||
check(ws.send("""["EVENT",${OptimizedJsonMapper.toJson(e)}]""")) {
|
||||
"publish to strfry failed"
|
||||
}
|
||||
}
|
||||
var oks = 0
|
||||
withTimeout(30_000) {
|
||||
while (oks < events.size) {
|
||||
if (incoming.receive().startsWith("[\"OK\"")) oks++
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
ws.close(1000, "preload-done")
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun geodeReconcilesAgainstStrfryRelay() =
|
||||
runBlocking {
|
||||
if (!enabled) {
|
||||
println("[skip] GeodeVsStrfryNegentropySyncTest — set STRFRY_BIN=/path/to/strfry to enable")
|
||||
return@runBlocking
|
||||
}
|
||||
val strfryWs = startStrfry()
|
||||
|
||||
// Same overlap shape as GeodeVsGeodeNegentropySyncTest so
|
||||
// results are directly comparable: A=[0..14], local=[5..19].
|
||||
val all = makeEvents(20)
|
||||
val strfryEvents = all.subList(0, 15)
|
||||
val localEvents = all.subList(5, 20)
|
||||
|
||||
publishToStrfry(strfryWs, strfryEvents)
|
||||
|
||||
val filter = Filter(kinds = listOf(1))
|
||||
|
||||
// The wire we care about: kmp-negentropy (client) talking
|
||||
// to strfry's `Negentropy ne(storage, 500'000)` (server).
|
||||
// Symmetric difference must match the Geode-vs-Geode case.
|
||||
val res = InteropSyncDriver(httpClient).negotiate(strfryWs, filter, localEvents)
|
||||
assertNull(res.error, "Geode↔strfry NEG must not error: ${res.error}")
|
||||
assertEquals(
|
||||
strfryEvents.subList(0, 5).map { it.id }.toSet(),
|
||||
res.needIds,
|
||||
"client should NEED [0..4] from strfry",
|
||||
)
|
||||
assertEquals(
|
||||
localEvents.subList(10, 15).map { it.id }.toSet(),
|
||||
res.haveIds,
|
||||
"client should announce HAVE for [15..19]",
|
||||
)
|
||||
assertTrue(res.rounds <= 16, "expected ≤16 NEG-MSG rounds, got ${res.rounds}")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,191 @@
|
||||
/*
|
||||
* 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.geode.interop
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
import okhttp3.Response
|
||||
import okhttp3.WebSocket
|
||||
import okhttp3.WebSocketListener
|
||||
import kotlin.test.fail
|
||||
|
||||
/**
|
||||
* Equivalent of strfry's `test/syncTest.pl` driver for our interop
|
||||
* tests. Drives one round of NIP-77 *negotiation* (NEG-OPEN /
|
||||
* NEG-MSG / NEG-CLOSE) against a real `ws://` endpoint — a Geode
|
||||
* `LocalRelayServer`, a strfry process, or any other NIP-77 relay.
|
||||
*
|
||||
* The driver opens a raw WebSocket — no `NostrClient` overhead —
|
||||
* to keep the wire format under direct control, the same way
|
||||
* `strfry sync` does. That way we exercise the server's NEG-OPEN /
|
||||
* NEG-MSG / NEG-CLOSE handling with no client-side framing or
|
||||
* filter-management indirection.
|
||||
*
|
||||
* Note: this driver only computes the symmetric difference. The
|
||||
* actual *sync* (REQ for `needIds`, EVENT for `haveIds`) is the
|
||||
* caller's job; that's a NIP-01 follow-up, not part of NIP-77.
|
||||
*/
|
||||
class InteropSyncDriver(
|
||||
private val httpClient: OkHttpClient = OkHttpClient.Builder().build(),
|
||||
) {
|
||||
/**
|
||||
* Negotiates the symmetric difference between `localEvents` and
|
||||
* the relay at [wsUrl] under [filter]. Returns the id sets so
|
||||
* the caller can close the loop with REQ / EVENT.
|
||||
*
|
||||
* @param wsUrl the source relay's `ws://…` URL.
|
||||
* @param filter NEG-OPEN filter — usually the broadest filter the
|
||||
* sync should cover (e.g. `Filter(kinds = listOf(1))`).
|
||||
* @param localEvents events the caller already has; the relay
|
||||
* reconciles these against its own snapshot.
|
||||
* @param frameSizeLimit `0` lets the relay choose. We pass `0`
|
||||
* here so the relay's configured cap (500_000 by default) is
|
||||
* what governs framing — same shape as `strfry sync`.
|
||||
* @param timeoutMs hard timeout on a single NEG-MSG round trip.
|
||||
* @param maxRounds upper bound on round trips. Strfry typically
|
||||
* converges in ≤5 rounds for 100 k corpora; 64 is a generous
|
||||
* safety net that catches pathological splits without hanging
|
||||
* tests forever.
|
||||
*/
|
||||
suspend fun negotiate(
|
||||
wsUrl: String,
|
||||
filter: Filter,
|
||||
localEvents: List<Event>,
|
||||
subId: String = "interop-sync",
|
||||
frameSizeLimit: Long = 0,
|
||||
timeoutMs: Long = 30_000L,
|
||||
maxRounds: Int = 64,
|
||||
): Result {
|
||||
val incoming = Channel<String>(UNLIMITED)
|
||||
val ws =
|
||||
httpClient.newWebSocket(
|
||||
Request.Builder().url(wsUrl.replace("ws://", "http://")).build(),
|
||||
object : WebSocketListener() {
|
||||
override fun onMessage(
|
||||
webSocket: WebSocket,
|
||||
text: String,
|
||||
) {
|
||||
incoming.trySend(text)
|
||||
}
|
||||
|
||||
override fun onClosing(
|
||||
webSocket: WebSocket,
|
||||
code: Int,
|
||||
reason: String,
|
||||
) {
|
||||
incoming.close()
|
||||
}
|
||||
|
||||
override fun onFailure(
|
||||
webSocket: WebSocket,
|
||||
t: Throwable,
|
||||
response: Response?,
|
||||
) {
|
||||
incoming.close(t)
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
return try {
|
||||
val session = NegentropySession(subId, filter, localEvents, frameSizeLimit)
|
||||
|
||||
// Step 1: NEG-OPEN.
|
||||
check(ws.send(OptimizedJsonMapper.toJson(session.open()))) { "send NEG-OPEN failed" }
|
||||
|
||||
// Step 2: drive NEG-MSG round trips until the client-side
|
||||
// session reports completion.
|
||||
val haveIds = mutableSetOf<HexKey>()
|
||||
val needIds = mutableSetOf<HexKey>()
|
||||
var rounds = 0
|
||||
while (rounds < maxRounds) {
|
||||
val raw = withTimeout(timeoutMs) { incoming.receive() }
|
||||
val msg = OptimizedJsonMapper.fromJsonToMessage(raw)
|
||||
rounds++
|
||||
when (msg) {
|
||||
is NegErrMessage -> {
|
||||
return Result(
|
||||
haveIds = haveIds,
|
||||
needIds = needIds,
|
||||
rounds = rounds,
|
||||
error = "${msg.subId}: ${msg.reason}",
|
||||
)
|
||||
}
|
||||
|
||||
is NoticeMessage -> {
|
||||
return Result(
|
||||
haveIds = haveIds,
|
||||
needIds = needIds,
|
||||
rounds = rounds,
|
||||
error = "NOTICE: ${msg.message}",
|
||||
)
|
||||
}
|
||||
|
||||
is NegMsgMessage -> {
|
||||
val r = session.processMessage(msg.message)
|
||||
haveIds += r.haveIds
|
||||
needIds += r.needIds
|
||||
if (r.isComplete()) {
|
||||
return Result(haveIds, needIds, rounds, error = null)
|
||||
}
|
||||
check(ws.send(OptimizedJsonMapper.toJson(r.nextCmd!!))) {
|
||||
"send NEG-MSG failed"
|
||||
}
|
||||
}
|
||||
|
||||
else -> {
|
||||
fail("unexpected message during NEG sync: ${msg::class.simpleName}")
|
||||
}
|
||||
}
|
||||
}
|
||||
Result(haveIds, needIds, rounds, error = "did not converge in $maxRounds rounds")
|
||||
} finally {
|
||||
ws.close(1000, "interop-test-done")
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of a reconciliation. `error` is non-null on
|
||||
* NEG-ERR/NOTICE/timeout; otherwise the id-set fields are
|
||||
* authoritative.
|
||||
*
|
||||
* @param haveIds events the client (us) had that the relay did not.
|
||||
* @param needIds events the relay had that the client (us) lacked.
|
||||
* @param rounds NEG-MSG round trips, including the one carrying
|
||||
* the terminator.
|
||||
*/
|
||||
data class Result(
|
||||
val haveIds: Set<HexKey>,
|
||||
val needIds: Set<HexKey>,
|
||||
val rounds: Int,
|
||||
val error: String?,
|
||||
)
|
||||
}
|
||||
+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.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||
|
||||
/**
|
||||
* 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 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(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.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.channels.BufferOverflow
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
@@ -161,4 +162,19 @@ class LiveEventStore(
|
||||
}
|
||||
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
|
||||
|
||||
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.nip77Negentropy.NegCloseCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropyServerSession
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||
|
||||
/**
|
||||
* 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
|
||||
* from [RelaySession.receive] — that path is single-threaded per the
|
||||
* 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(
|
||||
private val store: LiveEventStore,
|
||||
private val send: (Message) -> Unit,
|
||||
private val settings: NegentropySettings = NegentropySettings.Default,
|
||||
) {
|
||||
private val sessions = HashMap<String, NegentropyServerSession>()
|
||||
|
||||
/**
|
||||
* Open a reconciliation session. The relay snapshots its matching
|
||||
* events at this instant — concurrent inserts during the sync are
|
||||
* not surfaced; clients re-open if they want fresh state.
|
||||
* Open a reconciliation session. The relay snapshots the matching
|
||||
* `(created_at, id)` pairs at this instant — concurrent inserts
|
||||
* 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
|
||||
* AUTH or has kind/pubkey allow-deny lists applies the same rules
|
||||
* 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(
|
||||
cmd: NegOpenCmd,
|
||||
@@ -66,20 +84,43 @@ class NegSessionRegistry(
|
||||
}
|
||||
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.
|
||||
sessions.remove(cmd.subId)
|
||||
|
||||
val events = store.snapshotQuery(filters)
|
||||
val session = NegentropyServerSession(cmd.subId, events)
|
||||
val cap = settings.maxSyncEvents
|
||||
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
|
||||
|
||||
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) {
|
||||
val session = sessions[cmd.subId]
|
||||
if (session == null) {
|
||||
send(NegErrMessage(cmd.subId, "error: no negentropy session for ${cmd.subId}"))
|
||||
send(NegErrMessage(cmd.subId, "closed: unknown subscription handle"))
|
||||
return
|
||||
}
|
||||
runMessage(cmd.subId, session) { it.processMessage(cmd.message) }
|
||||
@@ -99,6 +140,9 @@ class NegSessionRegistry(
|
||||
sessions.clear()
|
||||
}
|
||||
|
||||
/** Test/diagnostic accessor. */
|
||||
val activeSessionCount: Int get() = sessions.size
|
||||
|
||||
private inline fun runMessage(
|
||||
subId: String,
|
||||
session: NegentropyServerSession,
|
||||
@@ -107,9 +151,11 @@ class NegSessionRegistry(
|
||||
try {
|
||||
val response = block(session)
|
||||
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)
|
||||
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.relay.server.policies.VerifyPolicy
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||
import com.vitorpamplona.quartz.utils.cache.LargeCache
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
@@ -43,12 +44,16 @@ import kotlin.coroutines.CoroutineContext
|
||||
* coroutine inside [VerifyPolicy]. Callers that flip this on should
|
||||
* *omit* `VerifyPolicy` from their [policyBuilder] chain to avoid
|
||||
* 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(
|
||||
private val store: IEventStore,
|
||||
private val policyBuilder: () -> IRelayPolicy = { VerifyPolicy },
|
||||
private val parentContext: CoroutineContext = SupervisorJob(),
|
||||
parallelVerify: Boolean = false,
|
||||
private val negentropySettings: NegentropySettings = NegentropySettings.Default,
|
||||
) : AutoCloseable {
|
||||
/** Scope for all subscriptions. */
|
||||
private val scope = CoroutineScope(parentContext + SupervisorJob())
|
||||
@@ -87,6 +92,7 @@ class NostrServer(
|
||||
onClose = { session ->
|
||||
connections.remove(session.hashCode())
|
||||
},
|
||||
negentropySettings = negentropySettings,
|
||||
).also { 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.NegMsgCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd
|
||||
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
|
||||
import com.vitorpamplona.quartz.utils.Log
|
||||
import com.vitorpamplona.quartz.utils.cache.LargeCache
|
||||
import kotlinx.coroutines.CancellationException
|
||||
@@ -56,11 +57,12 @@ class RelaySession(
|
||||
private val scope: CoroutineScope,
|
||||
private val onSend: (String) -> Unit,
|
||||
private val onClose: (RelaySession) -> Unit,
|
||||
negentropySettings: NegentropySettings = NegentropySettings.Default,
|
||||
) : AutoCloseable {
|
||||
private val subscriptions = LargeCache<String, Job>()
|
||||
|
||||
/** NIP-77 negentropy state for this connection. */
|
||||
private val negentropy = NegSessionRegistry(store, ::send)
|
||||
private val negentropy = NegSessionRegistry(store, ::send, negentropySettings)
|
||||
|
||||
private fun addSubscription(
|
||||
subId: String,
|
||||
|
||||
@@ -95,6 +95,38 @@ interface IEventStore : AutoCloseable {
|
||||
|
||||
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(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 snapshotIdsForNegentropy(
|
||||
filters: List<Filter>,
|
||||
maxEntries: Int?,
|
||||
): List<IdAndTime> = inner.snapshotIdsForNegentropy(filters, maxEntries)
|
||||
|
||||
override suspend fun delete(filter: Filter) {
|
||||
inner.delete(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.normalizeRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||
|
||||
/**
|
||||
* 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 snapshotIdsForNegentropy(
|
||||
filters: List<Filter>,
|
||||
maxEntries: Int?,
|
||||
): List<IdAndTime> = store.snapshotIdsForNegentropy(filters, maxEntries)
|
||||
|
||||
override suspend fun delete(filter: 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.isAddressable
|
||||
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.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 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.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IEventStore
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||
import com.vitorpamplona.quartz.nip40Expiration.isExpired
|
||||
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 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(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.storage.StorageVector
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.store.IdAndTime
|
||||
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
|
||||
* 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:
|
||||
* 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
|
||||
* 3. Send back the resulting [NegMsgMessage]
|
||||
* 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(
|
||||
val subId: String,
|
||||
localEvents: List<Event>,
|
||||
frameSizeLimit: Long = 0,
|
||||
localEntries: List<IdAndTime>,
|
||||
frameSizeLimit: Long = DEFAULT_FRAME_SIZE_LIMIT,
|
||||
) {
|
||||
private val storage = StorageVector()
|
||||
private val negentropy: Negentropy
|
||||
|
||||
init {
|
||||
for (event in localEvents) {
|
||||
storage.insert(event.createdAt, event.id)
|
||||
for (entry in localEntries) {
|
||||
storage.insert(entry.createdAt, entry.id)
|
||||
}
|
||||
storage.seal()
|
||||
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? {
|
||||
val msgBytes = Hex.decode(hexMessage)
|
||||
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()
|
||||
}
|
||||
}
|
||||
+104
@@ -0,0 +1,104 @@
|
||||
/*
|
||||
* 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
|
||||
|
||||
/**
|
||||
* 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.
|
||||
// cap=10 with 30 rows → result must be the +1 sentinel
|
||||
// (11 rows). Caller compares `size > cap` to detect
|
||||
// overflow — matches strfry's `maxSyncEvents` idiom.
|
||||
val capped = db.snapshotIdsForNegentropy(listOf(filter), maxEntries = 10)
|
||||
assertEquals(11, capped.size)
|
||||
|
||||
// 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()
|
||||
|
||||
// Server processes via NegentropyServerSession
|
||||
val serverSession = NegentropyServerSession("sub1", serverEvents)
|
||||
val serverSession = NegentropyServerSession.fromEvents("sub1", serverEvents)
|
||||
val response = serverSession.processMessage(openCmd.initialMessage)
|
||||
|
||||
assertNotNull(response)
|
||||
|
||||
Reference in New Issue
Block a user