mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-06-05 21:49:31 +00:00
42e0aa43d1
* persistency: per-job SQLite-backed storage layer (singleton, brokered)
Adds a backend-neutral CRUD library at waku/persistency/, plus the
nim-brokers dependency swap that enables it.
Architecture (ports-and-adapters):
* Persistency: process-wide singleton, one root directory.
* Job: one tenant, one DB file, one worker thread, one BrokerContext.
* Backend: SQLite via waku/common/databases/db_sqlite. Uniform schema
kv(category BLOB, key BLOB, payload BLOB) PRIMARY KEY (category, key)
WITHOUT ROWID, WAL mode.
* Writes are fire-and-forget via EventBroker(mt) PersistEvent.
* Reads are async via five RequestBroker(mt) shapes (KvGet, KvExists,
KvScan, KvCount, KvDelete). Reads return Result[T, PersistencyError].
* One storage thread per job; tenants isolated by BrokerContext.
Public surface (waku/persistency/persistency.nim):
Persistency.instance(rootDir) / Persistency.instance() / Persistency.reset()
p.openJob(id) / p.closeJob(id) / p.dropJob(id) / p.close()
p.job(id) / p[id] / p.hasJob(id)
Writes (Job form & string-id form, fire-and-forget):
persist / persistPut / persistDelete / persistEncoded
Reads (Job form & string-id form, async Result):
get / exists / scan / scanPrefix / count / deleteAcked
Key & payload encoding (keys.nim, payload.nim):
* encodePart family + variadic key(...) / payload(...) macros +
single-value toKey / toPayload.
* Primitives: string and openArray[byte] are 2-byte BE length + bytes;
int{8..64} are sign-flipped 8-byte BE; uint{16..64} are 8-byte BE;
bool/byte/char are 1 byte; enums are int64(ord(v)).
* Generic encodePart[T: tuple | object] recurses through fields() so
any composite Nim type is encodable without ceremony.
* Stable across Nim/C compiler upgrades: no sizeof, no memcpy, no
cast on pointers, no host-endianness dependency.
* `rawKey(bytes)` + `persistPut(..., openArray[byte])` let callers
bypass the built-in encoder with their own format (CBOR, protobuf...).
Lifecycle:
* Persistency.new is private; Persistency.instance is the only public
constructor. Same rootDir is idempotent; conflicting rootDir is
peInvalidArgument. Persistency.reset for test/restart paths.
* openJob opens-or-creates the per-job SQLite file; an existing file
is reused with its data preserved.
* Teardown integration: Persistency.instance registers a Teardown
MultiRequestBroker provider that closes all jobs and clears the
singleton slot when Waku.stop() issues Teardown.request.
Internal layering:
types.nim pure value types (Key, KeyRange, KvRow, TxOp,
PersistencyError)
keys.nim encodePart primitives + key(...) macro
payload.nim toPayload + payload(...) macro
schema.nim CREATE TABLE + connection pragmas + user_version
backend_sqlite.nim KvBackend, applyOps (single source of write SQL),
getOne/existsOne/deleteOne, scanRange (asc/desc,
half-open ranges, open-ended stop), countRange
backend_comm.nim EventBroker(mt) PersistEvent + 5 RequestBroker(mt)
declarations; encodeErr/decodeErr boundary helpers
backend_thread.nim startStorageThread / stopStorageThread (shared
allocShared0 arg, cstring dbPath, atomic
ready/shutdown flags); per-thread provider
registration
persistency.nim Persistency + Job types, singleton state, public
facade
../requests/lifecycle_requests.nim
Teardown MultiRequestBroker
Tests (69 cases, all passing):
test_keys.nim sort-order invariants (length-prefix strings,
sign-flipped ints, composite tuples, prefix
range)
test_backend.nim round-trip / replace / delete-return-value /
batched atomicity / asc-desc-half-open-open-
ended scans / category isolation / batch
txDelete
test_lifecycle.nim open-or-create rootDir / non-dir collision /
reopen across sessions / idempotent openJob /
two-tenant parallel isolation / closeJob joins
worker / dropJob removes file / acked delete
test_facade.nim put-then-get / atomic batch / scanPrefix
asc/desc / deleteAcked hit-miss /
fire-and-forget delete / two-tenant facade
isolation
test_encoding.nim tuple/named-tuple/object keys, embedded Key,
enum encoding, field-major composite sort,
payload struct encoding, end-to-end struct
round-trip through SQLite
test_string_lookup.nim peJobNotFound semantics / hasJob / subscript /
persistPut+get via id / reads short-circuit /
writes drop+warn / persistEncoded via id /
scan parity Job-ref vs id
test_singleton.nim idempotent same-rootDir / different-rootDir
rejection / no-arg instance lifecycle / reset
retargets / reset idempotence / Teardown.request
end-to-end
Prerequisite delivered in the same series: replace the in-tree broker
implementation with the external nim-brokers package; update all
broker call-sites (waku_filter_v2, waku_relay, waku_rln_relay,
delivery_service, peer_manager, requests/*, factory/*, api tests, etc.)
to the new package API; chat2 made to compile again.
Note: SDS adapter (Phase 5 of the design) is deferred -- nim-sds is
still developed side-by-side and the persistency layer is intentionally
SDS-agnostic.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
* persistency: pin nim-brokers by URL+commit (workaround for stale registry)
The bare `brokers >= 2.0.1` form cannot resolve on machines where the
local nimble SAT solver enumerates only the registry-recorded 0.1.0 for
brokers. The nim-lang/packages entry for `brokers` carries no per-tag
metadata (only the URL), so until that registry entry is refreshed the
SAT solver clamps the available-versions list to 0.1.0 and rejects the
>= 2.0.1 constraint -- even though pkgs2 and pkgcache both have v2.0.1
cloned locally.
Pinning by URL+commit bypasses the registry path entirely. Inline
comment in waku.nimble documents the situation and the path back to
the bare form once nim-lang/packages is updated.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
* persistency: nph format pass
Run `nph` on all 57 Nim files touched by this PR. Pure formatting:
17 files re-styled, no semantic change. Suite still 69/69.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
* Fix build, add local-storage-path config, lazy init of Persistency from Waku start
* fix: fix nix deps
* fixes for nix build, regenerate deps
* reverting accidental dependency changes
* Fixing deps
* Apply suggestions from code review
Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com>
* persistency tests: migrate to suite / asyncTest / await
Match the in-tree test convention (procSuite -> suite, sync test +
waitFor -> asyncTest + await):
- procSuite "X": -> suite "X":
- For tests doing async work: test -> asyncTest, waitFor -> await.
- Poll helpers (proc waitFor(t: Job, ...) in test_lifecycle.nim,
proc waitUntilExists(...) in test_facade.nim and
test_string_lookup.nim) -> Future[bool] {.async.}, internal
`waitFor X` -> `await X`, internal `sleep(N)` ->
`await sleepAsync(chronos.milliseconds(N))`.
- Renamed test_lifecycle.nim's helper proc from `waitFor(t: Job, ...)`
-> `pollExists(t: Job, ...)`; the previous name shadowed
chronos.waitFor in the chronos macro expansion.
- `chronos.milliseconds(N)` explicitly qualified because `std/times`
also exports `milliseconds` (returning TimeInterval, not Duration).
- `check await x` -> `let okN = await x; check okN` to dodge chronos's
"yield in expr not lowered" with await-as-macro-argument.
- `(await x).foo()` -> `let awN = await x; ... awN.foo() ...` for the
same reason.
waku/persistency/persistency.nim: nph also pulled the proc signatures
across multiple lines; restored explicit `Future[void] {.async.}`
return types after the colon (an intermediate nph pass had elided them).
Suite: 71 / 71 OK against the new async write surface.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
* use idiomatic valueOr instead of ifs
* Reworked persistency shutdown, remove not necessary teardown mechanism
* Use const for DefaultStoragePath
* format to follow coding guidelines - no use of result and explicit returns - no functional change
---------
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com>
246 lines
7.5 KiB
Nim
246 lines
7.5 KiB
Nim
import
|
|
std/[net, options, strutils],
|
|
chronicles,
|
|
libp2p/crypto/crypto,
|
|
libp2p/multiaddress,
|
|
libp2p/crypto/curve25519,
|
|
libp2p/peerid,
|
|
secp256k1,
|
|
results
|
|
|
|
import
|
|
../waku_rln_relay/rln_relay,
|
|
../rest_api/endpoint/builder,
|
|
../discovery/waku_discv5,
|
|
../node/waku_metrics,
|
|
../common/logging,
|
|
../common/rate_limit/setting,
|
|
../waku_enr/capabilities,
|
|
./networks_config,
|
|
../waku_mix
|
|
|
|
export RlnRelayConf, RlnRelayCreds, RestServerConf, Discv5Conf, MetricsServerConf
|
|
|
|
logScope:
|
|
topics = "waku conf"
|
|
|
|
type WebSocketSecureConf* {.requiresInit.} = object
|
|
keyPath*: string
|
|
certPath*: string
|
|
|
|
type WebSocketConf* = object
|
|
port*: Port
|
|
secureConf*: Option[WebSocketSecureConf]
|
|
|
|
# TODO: should be defined in validator_signed.nim and imported here
|
|
type ProtectedShard* {.requiresInit.} = object
|
|
shard*: uint16
|
|
key*: secp256k1.SkPublicKey
|
|
|
|
type DnsDiscoveryConf* {.requiresInit.} = object
|
|
enrTreeUrl*: string
|
|
# TODO: should probably only have one set of name servers (see dnsaddrs)
|
|
nameServers*: seq[IpAddress]
|
|
|
|
type StoreSyncConf* {.requiresInit.} = object
|
|
rangeSec*: uint32
|
|
intervalSec*: uint32
|
|
relayJitterSec*: uint32
|
|
|
|
type MixConf* = ref object
|
|
mixKey*: Curve25519Key
|
|
mixPubKey*: Curve25519Key
|
|
mixnodes*: seq[MixNodePubInfo]
|
|
|
|
type KademliaDiscoveryConf* = object
|
|
bootstrapNodes*: seq[(PeerId, seq[MultiAddress])]
|
|
## Bootstrap nodes for extended kademlia discovery.
|
|
|
|
type StoreServiceConf* {.requiresInit.} = object
|
|
dbMigration*: bool
|
|
dbURl*: string
|
|
dbVacuum*: bool
|
|
maxNumDbConnections*: int
|
|
retentionPolicies*: seq[string]
|
|
resume*: bool
|
|
storeSyncConf*: Option[StoreSyncConf]
|
|
|
|
type FilterServiceConf* {.requiresInit.} = object
|
|
maxPeersToServe*: uint32
|
|
subscriptionTimeout*: uint16
|
|
maxCriteria*: uint32
|
|
|
|
type EndpointConf* = object # TODO: make enum
|
|
natStrategy*: string
|
|
p2pTcpPort*: Port
|
|
dns4DomainName*: Option[string]
|
|
p2pListenAddress*: IpAddress
|
|
extMultiAddrs*: seq[MultiAddress]
|
|
extMultiAddrsOnly*: bool
|
|
|
|
## `WakuConf` is a valid configuration for a Waku node
|
|
## All information needed by a waku node should be contained
|
|
## In this object. A convenient `validate` method enables doing
|
|
## sanity checks beyond type enforcement.
|
|
## If `Option` is `some` it means the related protocol is enabled.
|
|
type WakuConf* {.requiresInit.} = ref object
|
|
# ref because `getRunningNetConfig` modifies it
|
|
nodeKey*: crypto.PrivateKey
|
|
|
|
clusterId*: uint16
|
|
subscribeShards*: seq[uint16]
|
|
protectedShards*: seq[ProtectedShard]
|
|
|
|
shardingConf*: ShardingConf
|
|
contentTopics*: seq[string]
|
|
|
|
relay*: bool
|
|
lightPush*: bool
|
|
peerExchangeService*: bool
|
|
peerExchangeDiscovery*: bool
|
|
|
|
# TODO: remove relay peer exchange
|
|
relayPeerExchange*: bool
|
|
rendezvous*: bool
|
|
circuitRelayClient*: bool
|
|
|
|
discv5Conf*: Option[Discv5Conf]
|
|
dnsDiscoveryConf*: Option[DnsDiscoveryConf]
|
|
filterServiceConf*: Option[FilterServiceConf]
|
|
storeServiceConf*: Option[StoreServiceConf]
|
|
rlnRelayConf*: Option[RlnRelayConf]
|
|
restServerConf*: Option[RestServerConf]
|
|
metricsServerConf*: Option[MetricsServerConf]
|
|
webSocketConf*: Option[WebSocketConf]
|
|
mixConf*: Option[MixConf]
|
|
kademliaDiscoveryConf*: Option[KademliaDiscoveryConf]
|
|
|
|
portsShift*: uint16
|
|
dnsAddrsNameServers*: seq[IpAddress]
|
|
endpointConf*: EndpointConf
|
|
wakuFlags*: CapabilitiesBitfield
|
|
|
|
# TODO: could probably make it a `PeerRemoteInfo`
|
|
staticNodes*: seq[string]
|
|
remoteStoreNode*: Option[string]
|
|
remoteLightPushNode*: Option[string]
|
|
remoteFilterNode*: Option[string]
|
|
remotePeerExchangeNode*: Option[string]
|
|
|
|
maxMessageSizeBytes*: uint64
|
|
|
|
logLevel*: logging.LogLevel
|
|
logFormat*: logging.LogFormat
|
|
|
|
peerPersistence*: bool
|
|
# TODO: should clearly be a uint
|
|
peerStoreCapacity*: Option[int]
|
|
# TODO: should clearly be a uint
|
|
maxConnections*: int
|
|
|
|
agentString*: string
|
|
|
|
colocationLimit*: int
|
|
|
|
rateLimit*: ProtocolRateLimitSettings
|
|
|
|
# TODO: those could be in a relay conf object
|
|
maxRelayPeers*: Option[int]
|
|
relayShardedPeerManagement*: bool
|
|
# TODO: use proper type
|
|
relayServiceRatio*: string
|
|
|
|
p2pReliability*: bool
|
|
|
|
localStoragePath*: string
|
|
|
|
proc logConf*(conf: WakuConf) =
|
|
info "Configuration: Enabled protocols",
|
|
relay = conf.relay,
|
|
rlnRelay = conf.rlnRelayConf.isSome(),
|
|
store = conf.storeServiceConf.isSome(),
|
|
filter = conf.filterServiceConf.isSome(),
|
|
lightPush = conf.lightPush,
|
|
peerExchange = conf.peerExchangeService,
|
|
rendezvous = conf.rendezvous
|
|
|
|
info "Configuration. Network", cluster = conf.clusterId
|
|
|
|
for shard in conf.subscribeShards:
|
|
info "Configuration. Active Relay Shards", shard = shard
|
|
|
|
if conf.discv5Conf.isSome():
|
|
for i in conf.discv5Conf.get().bootstrapNodes:
|
|
info "Configuration. Bootstrap nodes", node = i.string
|
|
|
|
if conf.rlnRelayConf.isSome():
|
|
var rlnRelayConf = conf.rlnRelayConf.get()
|
|
if rlnRelayConf.dynamic:
|
|
info "Configuration. Validation",
|
|
mechanism = "onchain rln",
|
|
contract = rlnRelayConf.ethContractAddress.string,
|
|
maxMessageSize = conf.maxMessageSizeBytes,
|
|
rlnEpochSizeSec = rlnRelayConf.epochSizeSec,
|
|
rlnRelayUserMessageLimit = rlnRelayConf.userMessageLimit
|
|
|
|
proc validateNodeKey(wakuConf: WakuConf): Result[void, string] =
|
|
wakuConf.nodeKey.getPublicKey().isOkOr:
|
|
return err("nodekey param is invalid")
|
|
return ok()
|
|
|
|
proc validateNoEmptyStrings(wakuConf: WakuConf): Result[void, string] =
|
|
if wakuConf.endpointConf.dns4DomainName.isSome() and
|
|
isEmptyOrWhiteSpace(wakuConf.endpointConf.dns4DomainName.get().string):
|
|
return err("dns4-domain-name is an empty string, set it to none(string) instead")
|
|
|
|
if isEmptyOrWhiteSpace(wakuConf.relayServiceRatio):
|
|
return err("relay-service-ratio is an empty string")
|
|
|
|
for sn in wakuConf.staticNodes:
|
|
if isEmptyOrWhiteSpace(sn):
|
|
return err("staticnode contain an empty string")
|
|
|
|
if wakuConf.remoteStoreNode.isSome() and
|
|
isEmptyOrWhiteSpace(wakuConf.remoteStoreNode.get()):
|
|
return err("storenode is an empty string, set it to none(string) instead")
|
|
|
|
if wakuConf.remoteLightPushNode.isSome() and
|
|
isEmptyOrWhiteSpace(wakuConf.remoteLightPushNode.get()):
|
|
return err("lightpushnode is an empty string, set it to none(string) instead")
|
|
|
|
if wakuConf.remotePeerExchangeNode.isSome() and
|
|
isEmptyOrWhiteSpace(wakuConf.remotePeerExchangeNode.get()):
|
|
return err("peer-exchange-node is an empty string, set it to none(string) instead")
|
|
|
|
if wakuConf.remoteFilterNode.isSome() and
|
|
isEmptyOrWhiteSpace(wakuConf.remoteFilterNode.get()):
|
|
return err("filternode is an empty string, set it to none(string) instead")
|
|
|
|
if wakuConf.dnsDiscoveryConf.isSome() and
|
|
isEmptyOrWhiteSpace(wakuConf.dnsDiscoveryConf.get().enrTreeUrl):
|
|
return err("dns-discovery-url is an empty string")
|
|
|
|
# TODO: rln relay config should validate itself
|
|
if wakuConf.rlnRelayConf.isSome():
|
|
let rlnRelayConf = wakuConf.rlnRelayConf.get()
|
|
|
|
if rlnRelayConf.ethClientUrls.len == 0:
|
|
return err("rln-relay-eth-client-address is empty")
|
|
if isEmptyOrWhiteSpace(rlnRelayConf.ethContractAddress):
|
|
return err("rln-relay-eth-contract-address is an empty string")
|
|
|
|
if rlnRelayConf.creds.isSome():
|
|
let creds = rlnRelayConf.creds.get()
|
|
if isEmptyOrWhiteSpace(creds.path):
|
|
return err ("rln-relay-cred-path is an empty string")
|
|
if isEmptyOrWhiteSpace(creds.password):
|
|
return err ("rln-relay-cred-password is an empty string")
|
|
|
|
return ok()
|
|
|
|
proc validate*(wakuConf: WakuConf): Result[void, string] =
|
|
?wakuConf.validateNodeKey()
|
|
?wakuConf.shardingConf.validateShards(wakuConf.subscribeShards)
|
|
?wakuConf.validateNoEmptyStrings()
|
|
return ok()
|