diff --git a/logos_delivery/waku/factory/conf_builder/store_service_conf_builder.nim b/logos_delivery/waku/factory/conf_builder/store_service_conf_builder.nim index abb37e7b8..651b4e814 100644 --- a/logos_delivery/waku/factory/conf_builder/store_service_conf_builder.nim +++ b/logos_delivery/waku/factory/conf_builder/store_service_conf_builder.nim @@ -1,5 +1,5 @@ import std/[strutils, sequtils], chronicles, results, chronos -import ../waku_conf, ./store_sync_conf_builder +import ../waku_conf logScope: topics = "waku conf builder store service" @@ -24,10 +24,9 @@ type StoreServiceConfBuilder* = object maxNumDbConnections*: Opt[int] retentionPolicies*: seq[string] resume*: Opt[bool] - storeSyncConf*: StoreSyncConfBuilder proc init*(T: type StoreServiceConfBuilder): StoreServiceConfBuilder = - StoreServiceConfBuilder(storeSyncConf: StoreSyncConfBuilder.init()) + StoreServiceConfBuilder() proc withEnabled*(b: var StoreServiceConfBuilder, enabled: bool) = b.enabled = Opt.some(enabled) @@ -90,9 +89,6 @@ proc build*(b: StoreServiceConfBuilder): Result[Opt[StoreServiceConf], string] = if b.dbUrl.get("") == "": return err "store.dbUrl is not specified" - let storeSyncConf = b.storeSyncConf.build().valueOr: - return err("Store Sync Conf failed to build") - let retentionPolicies = if b.retentionPolicies.len == 0: @[DefaultStoreRetentionPolicy] @@ -110,7 +106,6 @@ proc build*(b: StoreServiceConfBuilder): Result[Opt[StoreServiceConf], string] = maxNumDbConnections: b.maxNumDbConnections.get(DefaultStoreMaxNumDbConnections), retentionPolicies: retentionPolicies, resume: b.resume.get(DefaultStoreResume), - storeSyncConf: storeSyncConf, ) ) ) diff --git a/logos_delivery/waku/factory/conf_builder/store_sync_conf_builder.nim b/logos_delivery/waku/factory/conf_builder/store_sync_conf_builder.nim index dca53ab71..607fd9e21 100644 --- a/logos_delivery/waku/factory/conf_builder/store_sync_conf_builder.nim +++ b/logos_delivery/waku/factory/conf_builder/store_sync_conf_builder.nim @@ -1,10 +1,12 @@ import chronicles, results -import ../waku_conf +import ../waku_conf, ../../common/databases/dburl logScope: topics = "waku conf builder store sync" -const DefaultStoreSyncEnabled: bool = false +const + DefaultStoreSyncEnabled: bool = false + DefaultStoreSyncDbUrl*: string = "sqlite://:memory:" ################################## ## Store Sync Config Builder ## @@ -15,6 +17,7 @@ type StoreSyncConfBuilder* = object rangeSec*: Opt[uint32] intervalSec*: Opt[uint32] relayJitterSec*: Opt[uint32] + dbUrl*: Opt[string] proc init*(T: type StoreSyncConfBuilder): StoreSyncConfBuilder = StoreSyncConfBuilder() @@ -31,6 +34,9 @@ proc withIntervalSec*(b: var StoreSyncConfBuilder, intervalSec: uint32) = proc withRelayJitterSec*(b: var StoreSyncConfBuilder, relayJitterSec: uint32) = b.relayJitterSec = Opt.some(relayJitterSec) +proc withDbUrl*(b: var StoreSyncConfBuilder, dbUrl: string) = + b.dbUrl = Opt.some(dbUrl) + proc build*(b: StoreSyncConfBuilder): Result[Opt[StoreSyncConf], string] = if not b.enabled.get(DefaultStoreSyncEnabled): return ok(Opt.none(StoreSyncConf)) @@ -42,12 +48,22 @@ proc build*(b: StoreSyncConfBuilder): Result[Opt[StoreSyncConf], string] = if b.relayJitterSec.isNone(): return err "store.relayJitterSec is not specified" + if b.rangeSec.get() == 0: + return err "store sync range must be greater than 0" + + let dbUrl = b.dbUrl.get(DefaultStoreSyncDbUrl) + let engine = getDbEngine(dbUrl).valueOr: + return err "store sync dbUrl is invalid: " & error + if engine != "sqlite" and engine != "postgres": + return err "store sync dbUrl engine must be sqlite or postgres, got: " & engine + return ok( Opt.some( StoreSyncConf( rangeSec: b.rangeSec.get(), intervalSec: b.intervalSec.get(), relayJitterSec: b.relayJitterSec.get(), + dbUrl: dbUrl, ) ) ) diff --git a/logos_delivery/waku/factory/conf_builder/waku_conf_builder.nim b/logos_delivery/waku/factory/conf_builder/waku_conf_builder.nim index 469506e55..6d254090c 100644 --- a/logos_delivery/waku/factory/conf_builder/waku_conf_builder.nim +++ b/logos_delivery/waku/factory/conf_builder/waku_conf_builder.nim @@ -49,7 +49,6 @@ const DefaultLightPush: bool = false DefaultPeerExchange: bool = false # historical confbuilder default; wakunode2 CLI deviates (true) - DefaultStoreSyncMount: bool = false DefaultRendezvous: bool = false # historical confbuilder default; wakunode2 CLI deviates (true) DefaultMix*: bool = false @@ -117,6 +116,7 @@ type WakuConfBuilder* = object restServerConf*: RestServerConfBuilder rlnRelayConf*: RlnConfBuilder storeServiceConf*: StoreServiceConfBuilder + storeSyncConf*: StoreSyncConfBuilder mixConf*: MixConfBuilder webSocketConf*: WebSocketConfBuilder quicConf*: QuicConfBuilder @@ -126,7 +126,6 @@ type WakuConfBuilder* = object relay: Opt[bool] lightPush: Opt[bool] peerExchange: Opt[bool] - storeSync: Opt[bool] relayPeerExchange: Opt[bool] mix: Opt[bool] @@ -181,6 +180,7 @@ proc init*(T: type WakuConfBuilder): WakuConfBuilder = restServerConf: RestServerConfBuilder.init(), rlnRelayConf: RlnConfBuilder.init(), storeServiceConf: StoreServiceConfBuilder.init(), + storeSyncConf: StoreSyncConfBuilder.init(), webSocketConf: WebSocketConfBuilder.init(), quicConf: QuicConfBuilder.init(), rateLimitConf: RateLimitConfBuilder.init(), @@ -221,9 +221,6 @@ proc withRelay*(b: var WakuConfBuilder, relay: bool) = proc withLightPush*(b: var WakuConfBuilder, lightPush: bool) = b.lightPush = Opt.some(lightPush) -proc withStoreSync*(b: var WakuConfBuilder, storeSync: bool) = - b.storeSync = Opt.some(storeSync) - proc withPeerExchange*(b: var WakuConfBuilder, peerExchange: bool) = b.peerExchange = Opt.some(peerExchange) @@ -566,13 +563,6 @@ proc build*( warn "whether to mount peerExchange is not specified, defaulting to not mounting" DefaultPeerExchange - let storeSync = - if builder.storeSync.isSome(): - builder.storeSync.get() - else: - warn "whether to mount storeSync is not specified, defaulting to not mounting" - DefaultStoreSyncMount - let rendezvous = if builder.rendezvous.isSome(): builder.rendezvous.get() @@ -643,6 +633,9 @@ proc build*( let storeServiceConf = builder.storeServiceConf.build().valueOr: return err("Store Conf building failed: " & $error) + let storeSyncConf = builder.storeSyncConf.build().valueOr: + return err("Store Sync Conf building failed: " & $error) + let mixConf = builder.mixConf.build().valueOr: return err("Mix Conf building failed: " & $error) @@ -759,7 +752,7 @@ proc build*( filter = filterServiceConf.isSome, store = storeServiceConf.isSome, relay = relay, - sync = storeServiceConf.isSome() and storeServiceConf.get().storeSyncConf.isSome, + sync = storeSyncConf.isSome, mix = mix, ) @@ -780,6 +773,7 @@ proc build*( let wakuConf = WakuConf( # confs storeServiceConf: storeServiceConf, + storeSyncConf: storeSyncConf, filterServiceConf: filterServiceConf, discv5Conf: discv5Conf, rlnRelayConf: rlnRelayConf, diff --git a/logos_delivery/waku/factory/node_factory.nim b/logos_delivery/waku/factory/node_factory.nim index 5cd06b457..891d2602a 100644 --- a/logos_delivery/waku/factory/node_factory.nim +++ b/logos_delivery/waku/factory/node_factory.nim @@ -25,6 +25,7 @@ import ../waku_core/codecs, ../rln, ../discovery/waku_dnsdisc, + ../common/error_handling, ../waku_archive/retention_policy as policy, ../waku_archive/retention_policy/builder as policy_builder, ../waku_archive/driver as driver, @@ -147,6 +148,30 @@ proc getAutoshards*( autoShards.add(shard) return ok(autoshards) +proc setupArchive( + node: WakuNode, + dbUrl: string, + dbVacuum: bool, + dbMigration: bool, + maxNumDbConnections: int, + retentionPolicies: seq[string], + onFatalErrorAction: OnFatalErrorHandler, +): Future[Result[void, string]] {.async.} = + let archiveDriver = ( + await driver.ArchiveDriver.new( + dbUrl, dbVacuum, dbMigration, maxNumDbConnections, onFatalErrorAction + ) + ).valueOr: + return err("failed to setup archive driver: " & error) + + let retPolicies = policy.RetentionPolicy.new(retentionPolicies).valueOr: + return err("failed to create retention policy: " & error) + + node.mountArchive(archiveDriver, retPolicies).isOkOr: + return err("failed to mount waku archive protocol: " & error) + + return ok() + proc setupProtocols( node: WakuNode, conf: WakuConf ): Future[Result[void, string]] {.async.} = @@ -197,19 +222,14 @@ proc setupProtocols( if conf.storeServiceConf.isSome(): let storeServiceConf = conf.storeServiceConf.get() - let archiveDriver = ( - await driver.ArchiveDriver.new( + ( + await node.setupArchive( storeServiceConf.dbUrl, storeServiceConf.dbVacuum, storeServiceConf.dbMigration, - storeServiceConf.maxNumDbConnections, onFatalErrorAction, + storeServiceConf.maxNumDbConnections, storeServiceConf.retentionPolicies, + onFatalErrorAction, ) - ).valueOr: - return err("failed to setup archive driver: " & error) - - let retPolicies = policy.RetentionPolicy.new(storeServiceConf.retentionPolicies).valueOr: - return err("failed to create retention policy: " & error) - - node.mountArchive(archiveDriver, retPolicies).isOkOr: - return err("failed to mount waku archive protocol: " & error) + ).isOkOr: + return err(error) # Store setup try: @@ -217,24 +237,41 @@ proc setupProtocols( except CatchableError: return err("failed to mount waku store protocol: " & getCurrentExceptionMsg()) - if storeServiceConf.storeSyncConf.isSome(): - let confStoreSync = storeServiceConf.storeSyncConf.get() + if conf.storeSyncConf.isSome(): + let confStoreSync = conf.storeSyncConf.get() + if conf.storeServiceConf.isNone(): + # No store service: back reconciliation & transfer with a small + # archive bounded to twice the sync window (pruned every 30 min, + # so it transiently holds up to 2 * range + 30 min of messages). ( - await node.mountStoreSync( - conf.clusterId, conf.subscribeShards, conf.contentTopics, - confStoreSync.rangeSec, confStoreSync.intervalSec, - confStoreSync.relayJitterSec, + await node.setupArchive( + dbUrl = confStoreSync.dbUrl, + dbVacuum = false, + dbMigration = true, + maxNumDbConnections = 4, + retentionPolicies = @["time:" & $(2'u64 * confStoreSync.rangeSec)], + onFatalErrorAction = onFatalErrorAction, ) ).isOkOr: - return err("failed to mount waku store sync protocol: " & $error) + return err(error) - if conf.remoteStoreNode.isSome(): - let storeNode = parsePeerInfo(conf.remoteStoreNode.get()).valueOr: - return err("failed to set node waku store-sync peer: " & error) + ( + await node.mountStoreSync( + conf.clusterId, conf.subscribeShards, conf.contentTopics, + confStoreSync.rangeSec, confStoreSync.intervalSec, confStoreSync.relayJitterSec, + ) + ).isOkOr: + return err("failed to mount waku store sync protocol: " & $error) - node.peerManager.addServicePeer(storeNode, WakuReconciliationCodec) - node.peerManager.addServicePeer(storeNode, WakuTransferCodec) + if conf.storeServiceConf.isSome() and conf.remoteStoreNode.isSome(): + let storeNode = parsePeerInfo(conf.remoteStoreNode.get()).valueOr: + return err("failed to set node waku store-sync peer: " & error) + + node.peerManager.addServicePeer(storeNode, WakuReconciliationCodec) + node.peerManager.addServicePeer(storeNode, WakuTransferCodec) + elif conf.remoteStoreNode.isSome(): + warn "storenode is not used by store-sync without the store service; sync peers are found via discovery" mountStoreClient(node) if conf.remoteStoreNode.isSome(): diff --git a/logos_delivery/waku/factory/waku_conf.nim b/logos_delivery/waku/factory/waku_conf.nim index 614b1318f..7374f12e9 100644 --- a/logos_delivery/waku/factory/waku_conf.nim +++ b/logos_delivery/waku/factory/waku_conf.nim @@ -55,6 +55,9 @@ type StoreSyncConf* {.requiresInit.} = object rangeSec*: uint32 intervalSec*: uint32 relayJitterSec*: uint32 + dbUrl*: string + ## Backs reconciliation & transfer with a small bounded archive + ## when the store service is not enabled. Ignored otherwise. type MixConf* = ref object mixKey*: Curve25519Key @@ -68,7 +71,6 @@ type StoreServiceConf* = object maxNumDbConnections*: int retentionPolicies*: seq[string] resume*: bool - storeSyncConf*: Opt[StoreSyncConf] type FilterServiceConf* {.requiresInit.} = object maxPeersToServe*: uint32 @@ -113,6 +115,7 @@ type WakuConf* {.requiresInit.} = ref object dnsDiscoveryConf*: Opt[DnsDiscoveryConf] filterServiceConf*: Opt[FilterServiceConf] storeServiceConf*: Opt[StoreServiceConf] + storeSyncConf*: Opt[StoreSyncConf] rlnRelayConf*: Opt[RlnConf] restServerConf*: Opt[RestServerConf] metricsServerConf*: Opt[MetricsServerConf] diff --git a/logos_delivery/waku/node/waku_node.nim b/logos_delivery/waku/node/waku_node.nim index b1448eb2a..31ed0cc68 100644 --- a/logos_delivery/waku/node/waku_node.nim +++ b/logos_delivery/waku/node/waku_node.nim @@ -370,6 +370,9 @@ proc mountStoreSync*( storeSyncInterval: uint32, storeSyncRelayJitter: uint32, ): Future[Result[void, string]] {.async.} = + if node.wakuArchive.isNil(): + return err("store sync requires a mounted archive") + let idsChannel = newAsyncQueue[(SyncID, PubsubTopic, ContentTopic)](0) let wantsChannel = newAsyncQueue[(PeerId)](0) let needsChannel = newAsyncQueue[(PeerId, WakuMessageHash)](0) diff --git a/tests/factory/test_all.nim b/tests/factory/test_all.nim index 916610340..e0113f5eb 100644 --- a/tests/factory/test_all.nim +++ b/tests/factory/test_all.nim @@ -1,3 +1,3 @@ {.used.} -import ./test_waku_conf, ./test_node_factory +import ./test_waku_conf, ./test_node_factory, ./test_store_sync_standalone diff --git a/tests/factory/test_store_sync_standalone.nim b/tests/factory/test_store_sync_standalone.nim new file mode 100644 index 000000000..4807f8756 --- /dev/null +++ b/tests/factory/test_store_sync_standalone.nim @@ -0,0 +1,84 @@ +{.used.} + +import results, testutils/unittests, chronos, libp2p/protocols/connectivity/relay/relay + +import + tests/testlib/[wakunode, wakucore], + logos_delivery/waku/[waku_node, waku_enr], + logos_delivery/waku/factory/[node_factory, conf_builder/conf_builder] + +suite "Store Sync standalone (no store service)": + asynctest "store-sync without store service mounts sync backed by a small archive": + var confBuilder = defaultTestWakuConfBuilder() + confBuilder.storeSyncConf.withEnabled(true) + confBuilder.storeSyncConf.withRangeSec(1800) + confBuilder.storeSyncConf.withIntervalSec(60) + confBuilder.storeSyncConf.withRelayJitterSec(20) + let conf = confBuilder.build().value + + check: + conf.storeSyncConf.isSome() + conf.storeServiceConf.isNone() + conf.wakuFlags.supportsCapability(Capabilities.Sync) + + let node = (await setupNode(conf, relay = Relay.new())).valueOr: + raiseAssert error + + check: + not node.isNil() + not node.wakuArchive.isNil() + not node.wakuStoreReconciliation.isNil() + not node.wakuStoreTransfer.isNil() + node.wakuStore.isNil() + + asynctest "store service with store-sync keeps the store-node shape": + var confBuilder = defaultTestWakuConfBuilder() + confBuilder.storeServiceConf.withEnabled(true) + confBuilder.storeServiceConf.withDbUrl("sqlite://:memory:") + confBuilder.storeSyncConf.withEnabled(true) + confBuilder.storeSyncConf.withRangeSec(3600) + confBuilder.storeSyncConf.withIntervalSec(300) + confBuilder.storeSyncConf.withRelayJitterSec(20) + let conf = confBuilder.build().value + + check: + conf.storeServiceConf.isSome() + conf.storeSyncConf.isSome() + conf.wakuFlags.supportsCapability(Capabilities.Sync) + conf.wakuFlags.supportsCapability(Capabilities.Store) + + let node = (await setupNode(conf, relay = Relay.new())).valueOr: + raiseAssert error + + check: + not node.isNil() + not node.wakuStore.isNil() + not node.wakuArchive.isNil() + not node.wakuStoreReconciliation.isNil() + not node.wakuStoreTransfer.isNil() + + test "store-sync disabled leaves conf empty and Sync capability unset": + let conf = defaultTestWakuConf() + + check: + conf.storeSyncConf.isNone() + not conf.wakuFlags.supportsCapability(Capabilities.Sync) + + test "store-sync rejects a zero sync range": + var confBuilder = defaultTestWakuConfBuilder() + confBuilder.storeSyncConf.withEnabled(true) + confBuilder.storeSyncConf.withRangeSec(0) + confBuilder.storeSyncConf.withIntervalSec(60) + confBuilder.storeSyncConf.withRelayJitterSec(20) + + check confBuilder.build().isErr() + + test "store-sync rejects an unsupported db engine": + var confBuilder = defaultTestWakuConfBuilder() + confBuilder.storeSyncConf.withEnabled(true) + confBuilder.storeSyncConf.withRangeSec(1800) + confBuilder.storeSyncConf.withIntervalSec(60) + confBuilder.storeSyncConf.withRelayJitterSec(20) + confBuilder.storeSyncConf.withDbUrl("bogus://nope") + + check confBuilder.build().isErr() diff --git a/tools/confutils/cli_args.nim b/tools/confutils/cli_args.nim index 1774f1445..2a0fb5012 100644 --- a/tools/confutils/cli_args.nim +++ b/tools/confutils/cli_args.nim @@ -37,7 +37,7 @@ import ./envvar as confEnvvarDefs, ./envvar_net as confEnvvarNet export confTomlDefs, confTomlNet, confEnvvarDefs, confEnvvarNet, ProtectedShard, - DefaultMaxWakuMessageSizeStr, DefaultAgentString + DefaultMaxWakuMessageSizeStr, DefaultAgentString, DefaultStoreSyncDbUrl logScope: topics = "waku cli args" @@ -428,6 +428,14 @@ hence would have reachability issues.""", name: "store-sync-relay-jitter" .}: uint32 + storeSyncDbUrl* {. + hidden, + desc: + "Database connection URL backing store sync when the store service is disabled.", + defaultValue: DefaultStoreSyncDbUrl, + name: "store-sync-db-url" + .}: string + ## Filter config filter* {. desc: "Enable filter protocol: true|false", defaultValue: true, name: "filter" @@ -1110,10 +1118,11 @@ proc toWakuConf*(n: WakuNodeConf): ConfResult[WakuConf] = if n.peerExchangeNode != "": b.withRemotePeerExchangeNode(n.peerExchangeNode) - b.storeServiceConf.storeSyncConf.withEnabled(n.storeSync) - b.storeServiceConf.storeSyncConf.withIntervalSec(n.storeSyncInterval) - b.storeServiceConf.storeSyncConf.withRangeSec(n.storeSyncRange) - b.storeServiceConf.storeSyncConf.withRelayJitterSec(n.storeSyncRelayJitter) + b.storeSyncConf.withEnabled(n.storeSync) + b.storeSyncConf.withIntervalSec(n.storeSyncInterval) + b.storeSyncConf.withRangeSec(n.storeSyncRange) + b.storeSyncConf.withRelayJitterSec(n.storeSyncRelayJitter) + b.storeSyncConf.withDbUrl(n.storeSyncDbUrl) if n.mix.isSome(): b.mixConf.withEnabled(n.mix.get())