logos-messaging/logos-delivery
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-delivery.git synced 2026-06-05 05:29:45 +00:00
Code Issues Packages Projects Releases Wiki Activity
logos-delivery/waku/factory/waku_conf.nim
NagyZoltanPeter 42e0aa43d1
feat: persistency (#3880) 
* 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>
2026-05-16 00:09:07 +02:00

246 lines
7.5 KiB
Nim
Raw Blame History

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()
Reference in New Issue View Git Blame Copy Permalink