diff --git a/.dockerignore b/.dockerignore index 247ac61ce..1497e83ce 100644 --- a/.dockerignore +++ b/.dockerignore @@ -7,3 +7,5 @@ /nimcache librln* **/vendor/* +/nimbledeps +/build diff --git a/logos_delivery/api/conf/logos_delivery_conf_json.nim b/logos_delivery/api/conf/logos_delivery_conf_json.nim index 6440a63ac..97abb83f3 100644 --- a/logos_delivery/api/conf/logos_delivery_conf_json.nim +++ b/logos_delivery/api/conf/logos_delivery_conf_json.nim @@ -17,6 +17,8 @@ const # a flat blob before the WakuNodeConf walker sees them (it would reject them). KeyReliabilityEnabled = "reliabilityenabled" KeyReliability = "reliability" + KeySendConfirmation = "sendconfirmation" + KeySendConfirmationName = "send-confirmation" proc parseMode(s: string): Result[LogosDeliveryMode, string] = case s.strip().toLowerAscii() @@ -52,7 +54,9 @@ proc parseFlatConf( ## `WakuNodeConf`. Full stack. Delete this proc and its call site to drop support. var messaging = MessagingClientConf() var reliabilityFields: Table[string, (string, JsonNode)] - for key in [KeyReliabilityEnabled, KeyReliability]: + for key in [ + KeyReliabilityEnabled, KeyReliability, KeySendConfirmation, KeySendConfirmationName + ]: if topJsonNode.hasKey(key): reliabilityFields[key] = topJsonNode.getOrDefault(key) topJsonNode.del(key) diff --git a/logos_delivery/api/conf/messaging_conf.nim b/logos_delivery/api/conf/messaging_conf.nim index a8c91acd7..9d06b953e 100644 --- a/logos_delivery/api/conf/messaging_conf.nim +++ b/logos_delivery/api/conf/messaging_conf.nim @@ -38,6 +38,9 @@ type MessagingClientConf* = object ## RLN epoch size, in seconds. reliabilityEnabled* {.name: "reliability".}: Opt[bool] ## Enable store-based send reliability. + sendConfirmation* {.name: "send-confirmation".}: Opt[string] + ## How MessageSent is confirmed: "store" (store witness, default) or + ## "propagation" (publish-path peer count; no store dependency). store*: Opt[bool] ## Enable the store protocol. storenode* {.name: "storenode".}: Opt[string] storeMessageDbUrl* {.name: "store-message-db-url".}: Opt[string] diff --git a/logos_delivery/messaging/delivery_service/send_service/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim index 097bb0752..99ffeb7ea 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim @@ -38,6 +38,10 @@ const ArchiveTime = chronos.seconds(3) ## Estimation of the time we wait until we start confirming that a message has been properly ## received and archived by a store node +type SendConfirmationMode* {.pure.} = enum + Store ## MessageSent fires when a store node confirms the message (default) + Propagation ## MessageSent fires from the publish path (relay/lightpush peer count) + type SendService* = ref object of RootObj brokerCtx: BrokerContext taskCache: seq[DeliveryTask] @@ -48,6 +52,7 @@ type SendService* = ref object of RootObj sendProcessor: BaseSendProcessor waku: Waku + confirmationMode: SendConfirmationMode checkStoreForMessages: bool lastStoreCheckTime: Moment ## throttles store validation queries to ArchiveTime cadence @@ -77,14 +82,19 @@ proc setupSendProcessorChain( return ok(processors[0]) proc new*( - T: typedesc[SendService], preferP2PReliability: bool, waku: Waku + T: typedesc[SendService], + preferP2PReliability: bool, + waku: Waku, + confirmationMode = SendConfirmationMode.Store, ): Result[T, string] = if not waku.hasRelay() and not waku.hasLightpush(): return err( "Could not create SendService. wakuRelay or wakuLightpushClient should be set" ) - let checkStoreForMessages = preferP2PReliability and waku.isStoreMounted() + let checkStoreForMessages = + confirmationMode == SendConfirmationMode.Store and preferP2PReliability and + waku.isStoreMounted() let sendProcessorChain = setupSendProcessorChain(waku, waku.brokerCtx).valueOr: return err("failed to setup SendProcessorChain: " & $error) @@ -95,6 +105,7 @@ proc new*( serviceLoopHandle: nil, sendProcessor: sendProcessorChain, waku: waku, + confirmationMode: confirmationMode, checkStoreForMessages: checkStoreForMessages, lastStoreCheckTime: Moment.now(), ) @@ -170,6 +181,13 @@ proc reportTaskResult(self: SendService, task: DeliveryTask) = self.brokerCtx, task.requestId, task.msgHash.to0xHex() ) task.propagateEventEmitted = true + if self.confirmationMode == SendConfirmationMode.Propagation: + # The publish path is the confirmation signal: report sent right after + # propagation; the validated task is dropped by evaluateAndCleanUp. + info "Message successfully sent (propagation confirmed)", + requestId = task.requestId, msgHash = task.msgHash.to0xHex() + MessageSentEvent.emit(self.brokerCtx, task.requestId, task.msgHash.to0xHex()) + task.state = DeliveryState.SuccessfullyValidated return of DeliveryState.SuccessfullyValidated: info "Message successfully sent", @@ -276,5 +294,8 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} = await self.sendProcessor.process(task) reportTaskResult(self, task) - if task.state != DeliveryState.FailedToDeliver: + # finalized tasks (failed, or already confirmed in propagation mode) must + # not enter the cache: the service loop would report them a second time + if task.state notin + {DeliveryState.FailedToDeliver, DeliveryState.SuccessfullyValidated}: self.addTask(task) diff --git a/logos_delivery/messaging/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim index d883878fb..e7c9bd37c 100644 --- a/logos_delivery/messaging/messaging_client.nim +++ b/logos_delivery/messaging/messaging_client.nim @@ -1,6 +1,7 @@ ## Messaging layer core: the `MessagingClient` type plus its construction and ## lifecycle. The public operations (subscribe / unsubscribe / send) live in ## `messaging/api.nim`. +import std/strutils import results, chronos, chronicles import logos_delivery/api/conf/messaging_conf, @@ -24,7 +25,17 @@ proc new*( ## The messaging layer chains onto Waku: it drives the underlying Waku kernel ## for transport while exposing its own send/recv API. let reliability = conf.reliabilityEnabled.get(DefaultP2pReliability) - let sendService = ?SendService.new(reliability, waku) + + let confirmationMode = + case conf.sendConfirmation.get("store").strip().toLowerAscii() + of "store": + SendConfirmationMode.Store + of "propagation": + SendConfirmationMode.Propagation + else: + return err("invalid send-confirmation mode: " & conf.sendConfirmation.get("")) + + let sendService = ?SendService.new(reliability, waku, confirmationMode) let recvService = RecvService.new(waku) return ok( T( 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..6eb6ad011 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(): @@ -242,7 +279,10 @@ proc setupProtocols( return err("failed to set node waku store peer: " & error) node.peerManager.addServicePeer(storeNode, WakuStoreCodec) - if conf.storeServiceConf.isSome and conf.storeServiceConf.get().resume: + if (conf.storeServiceConf.isSome and conf.storeServiceConf.get().resume) or + conf.storeSyncConf.isSome(): + # sync-enabled full nodes track last-online so the startup catch-up can + # choose between neighbour reconciliation and store resume node.setupStoreResume() if conf.shardingConf.kind == AutoSharding: 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..3b10d5b7b 100644 --- a/logos_delivery/waku/node/waku_node.nim +++ b/logos_delivery/waku/node/waku_node.nim @@ -136,6 +136,8 @@ type wakuKademlia*: WakuKademlia ports*: BoundPorts relayReconnectFut*: Future[void] + storeCatchUpFut*: Future[void] + storeSyncRange*: timer.Duration SubscriptionManager* = ref object of RootObj node*: WakuNode @@ -370,12 +372,17 @@ 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) let pubsubTopics = shards.mapIt($RelayShard(clusterId: cluster, shardId: it)) + node.storeSyncRange = storeSyncRange.seconds + let recon = ?await SyncReconciliation.new( pubsubTopics, contentTopics, node.peerManager, node.wakuArchive, storeSyncRange.seconds, storeSyncInterval.seconds, storeSyncRelayJitter.seconds, @@ -391,8 +398,35 @@ proc mountStoreSync*( reconMountRes.isOkOr: return err(error.msg) + let transferValidator: TransferValidator = proc( + msg: WakuMessage + ): Future[bool] {.async.} = + ## RLN mounts after store sync (but before the switch starts accepting + ## connections), so capture the node and check at call time; nil rln + ## means RLN is not configured on this network. Freshness is not + ## checked: synced messages are old by design. Known gaps: archives do + ## not persist RLN proofs, so archive-served messages arrive proofless + ## and can only be counted, not verified (enforcement needs proof + ## persistence); and proofs built against roots older than the + ## acceptable root window are rejected, bounding history sync across + ## heavy membership churn. + if node.rln.isNil(): + return true + + if msg.proof.len == 0: + total_transfer_messages_unverified.inc() + return true + + let res = await node.rln.validateMessage(msg, checkFreshness = false) + return res == MessageValidationResult.Valid + let transfer = SyncTransfer.new( - node.peerManager, node.wakuArchive, idsChannel, wantsChannel, needsChannel + node.peerManager, + node.wakuArchive, + idsChannel, + wantsChannel, + needsChannel, + msgValidator = Opt.some(transferValidator), ) node.wakuStoreTransfer = transfer @@ -590,6 +624,66 @@ proc stopProvidersAndListeners*(node: WakuNode) = RequestContentTopicsHealth.clearProvider(node.brokerCtx) RequestShardTopicsHealth.clearProvider(node.brokerCtx) +func useSyncCatchUp*( + hasReconciliation: bool, + lastOnline: Timestamp, + now: Timestamp, + syncRange: timer.Duration, +): bool = + ## Startup catch-up decision: inside the sync window neighbour + ## reconciliation repairs the gap; beyond it (or on fresh start, when no + ## last-online timestamp exists) store resume is needed. + hasReconciliation and lastOnline > 0 and now - lastOnline <= syncRange.nanos + +proc storeStartupCatchUp(node: WakuNode) {.async.} = + ## Store as a startup-only dependency: estimate the offline gap from the + ## persisted last-online timestamp and catch up accordingly. Runs in the + ## background; node startup must not block on store or sync peers. + let lastOnline = node.wakuStoreResume.getLastOnlineTimestamp().valueOr: + error "catch-up: failed to read last online timestamp", error = error + Timestamp(0) + + let now = getNowInNanosecondTime() + + if useSyncCatchUp( + not node.wakuStoreReconciliation.isNil(), lastOnline, now, node.storeSyncRange + ): + info "offline gap within sync window, catching up via reconciliation", + gapSeconds = (now - lastOnline) div 1_000_000_000 + # sleep first: the switch may not be started yet and discovery may still + # be warming up; retry failed attempts, the periodic sync loop remains + # the safety net if none succeeds within the budget + for _ in 0 ..< 24: + await sleepAsync(5.seconds) + if node.peerManager.selectPeer(WakuReconciliationCodec).isNone(): + continue + (await node.wakuStoreReconciliation.storeSynchronization()).isOkOr: + error "startup reconciliation attempt failed", error = error + continue + return + warn "startup reconciliation did not complete; periodic sync remains the safety net" + else: + info "offline gap beyond sync window or fresh start, using store resume", + gapSeconds = (now - lastOnline) div 1_000_000_000 + # don't burn resume tries against an empty peer store: wait (bounded) + # for a store peer to be discovered first + var peerWait = 24 + var tries = 3 + while tries > 0: + if node.peerManager.selectPeer(WakuStoreCodec).isNone(): + peerWait -= 1 + if peerWait <= 0: + warn "no store peer found for startup resume" + return + await sleepAsync(5.seconds) + continue + (await node.wakuStoreResume.autoStoreResume()).isOkOr: + tries -= 1 + error "store resume failed", triesLeft = tries, error = $error + await sleepAsync(30.seconds) + continue + break + proc start*(node: WakuNode) {.async.} = ## Starts a created Waku Node and ## all its mounted protocols. @@ -603,7 +697,9 @@ proc start*(node: WakuNode) {.async.} = zeroPortPresent = true if not node.wakuStoreResume.isNil(): - await node.wakuStoreResume.start() + # catch up in the background: startup must not block on store peers + node.storeCatchUpFut = node.storeStartupCatchUp() + node.wakuStoreResume.startPeriodicOnly() if not node.wakuRendezvousClient.isNil(): await node.wakuRendezvousClient.start() @@ -676,6 +772,12 @@ proc stop*(node: WakuNode) {.async.} = if not node.wakuArchive.isNil(): await node.wakuArchive.stopWait() + if not node.storeCatchUpFut.isNil(): + # in-flight store queries wrap awaits in results.catch, which can swallow + # the one cancellation delivery; bound shutdown instead of waiting for + # their natural completion + discard await node.storeCatchUpFut.cancelAndWait().withTimeout(5.seconds) + if not node.wakuStoreResume.isNil(): await node.wakuStoreResume.stopWait() diff --git a/logos_delivery/waku/node/waku_node/store.nim b/logos_delivery/waku/node/waku_node/store.nim index 92d236a37..dca89d5df 100644 --- a/logos_delivery/waku/node/waku_node/store.nim +++ b/logos_delivery/waku/node/waku_node/store.nim @@ -114,6 +114,19 @@ proc mountStore*( node.switch.mount(node.wakuStore, protocolMatcher(store_common.WakuStoreCodec)) +proc queryArchive*( + node: WakuNode, request: StoreQueryRequest +): Future[StoreQueryResult] {.async.} = + ## Serves a store query directly from the local archive, for nodes that + ## carry an archive without mounting the store protocol (e.g. full nodes + ## running store sync standalone). Same conversion path as mountStore's + ## request handler, so pagination/cursor semantics are identical. + if node.wakuArchive.isNil(): + return err(StoreError.new(300, "waku archive is not available")) + + let response = await node.wakuArchive.findMessages(request.toArchiveQuery()) + return response.toStoreResult() + proc mountStoreClient*(node: WakuNode) = info "mounting store client" diff --git a/logos_delivery/waku/rest_api/endpoint/store/handlers.nim b/logos_delivery/waku/rest_api/endpoint/store/handlers.nim index aa0136922..deee32072 100644 --- a/logos_delivery/waku/rest_api/endpoint/store/handlers.nim +++ b/logos_delivery/waku/rest_api/endpoint/store/handlers.nim @@ -179,6 +179,23 @@ proc retrieveMsgsFromSelfNode( return resp +proc retrieveMsgsFromArchive( + self: WakuNode, storeQuery: StoreQueryRequest +): Future[RestApiResponse] {.async.} = + ## Serves a local store query from the node's archive when the store + ## protocol is not mounted (e.g. store-sync standalone full nodes). + + let storeResp = (await self.queryArchive(storeQuery)).valueOr: + return RestApiResponse.internalServerError($error) + + let resp = RestApiResponse.jsonResponse(storeResp.toHex(), status = Http200).valueOr: + const msg = "Error building the json respose" + let e = $error + error msg, error = e + return RestApiResponse.internalServerError(fmt("{msg} [{e}]")) + + return resp + # Subscribes the rest handler to attend "/store/v1/messages" requests proc installStoreApiHandlers*( router: var RestRouter, @@ -226,6 +243,13 @@ proc installStoreApiHandlers*( ## the local/self store node. return await node.retrieveMsgsFromSelfNode(storeQuery) + if peer.isNone() and not node.wakuArchive.isNil(): + ## No peer given and no store protocol, but the node carries an archive + ## (store-sync standalone): serve the query from the local archive, like + ## a store node serving its own. Only the archive's retention window is + ## returned; pass peerAddr to query a remote store for deeper history. + return await node.retrieveMsgsFromArchive(storeQuery) + # Parse the peer address parameter let parsedPeerAddr = parseUrlPeerAddr(peer).valueOr: return RestApiResponse.badRequest(error) diff --git a/logos_delivery/waku/rln/rln.nim b/logos_delivery/waku/rln/rln.nim index 633d433dd..3933f2aac 100644 --- a/logos_delivery/waku/rln/rln.nim +++ b/logos_delivery/waku/rln/rln.nim @@ -48,7 +48,7 @@ proc stop*(rlnPeer: Rln) {.async: (raises: [Exception]).} = await rlnPeer.groupManager.stop() proc validateMessage*( - rlnPeer: Rln, msg: WakuMessage + rlnPeer: Rln, msg: WakuMessage, checkFreshness = true ): Future[MessageValidationResult] {.async.} = ## validate the supplied `msg` based on the waku-rln-relay routing protocol i.e., ## the `msg`'s epoch is within MaxEpochGap of the current epoch @@ -56,6 +56,9 @@ proc validateMessage*( ## the `msg` does not violate the rate limit ## `timeOption` indicates Unix epoch time (fractional part holds sub-seconds) ## if `timeOption` is supplied, then the current epoch is calculated based on that + ## `checkFreshness = false` skips the timestamp-recency bound: used for + ## messages that are legitimately old, e.g. received via store sync + ## transfer. Proof, root and timestamp/epoch binding are still verified. let proof = RateLimitProof.init(msg.proof).valueOr: return MessageValidationResult.Invalid @@ -72,7 +75,7 @@ proc validateMessage*( info "time info", currentTime = currentTime, messageTime = messageTime, msgHash = msg.hash - if timeDiff > rlnPeer.rlnMaxTimestampGap: + if checkFreshness and timeDiff > rlnPeer.rlnMaxTimestampGap: warn "invalid message: timestamp difference exceeds threshold", timeDiff = timeDiff, maxTimestampGap = rlnPeer.rlnMaxTimestampGap, diff --git a/logos_delivery/waku/waku_store/resume.nim b/logos_delivery/waku/waku_store/resume.nim index 0f45b8a6b..4d380b30f 100644 --- a/logos_delivery/waku/waku_store/resume.nim +++ b/logos_delivery/waku/waku_store/resume.nim @@ -197,19 +197,10 @@ proc periodicSetLastOnline(self: StoreResume) {.async.} = self.setLastOnlineTimestamp(ts).isOkOr: error "failed to set last online timestamp", error, time = ts -proc start*(self: StoreResume) {.async.} = - # start resume process, will try thrice. - var tries = 3 - while tries > 0: - (await self.autoStoreResume()).isOkOr: - tries -= 1 - error "store resume failed", triesLeft = tries, error = $error - await sleepAsync(30.seconds) - continue - - break - - # starting periodic storage of last online timestamp +proc startPeriodicOnly*(self: StoreResume) = + ## Starts only the periodic last-online bookkeeping; the catch-up decision + ## (neighbour sync within the window vs store resume beyond it) is made by + ## the node's startup catch-up. self.handle = self.periodicSetLastOnline() proc stopWait*(self: StoreResume) {.async.} = diff --git a/logos_delivery/waku/waku_store_sync.nim b/logos_delivery/waku/waku_store_sync.nim index 03c1b33af..cadde72ea 100644 --- a/logos_delivery/waku/waku_store_sync.nim +++ b/logos_delivery/waku/waku_store_sync.nim @@ -1,6 +1,9 @@ {.push raises: [].} import - ./waku_store_sync/reconciliation, ./waku_store_sync/transfer, ./waku_store_sync/common + ./waku_store_sync/reconciliation, + ./waku_store_sync/transfer, + ./waku_store_sync/common, + ./waku_store_sync/protocols_metrics -export reconciliation, transfer, common +export reconciliation, transfer, common, protocols_metrics diff --git a/logos_delivery/waku/waku_store_sync/protocols_metrics.nim b/logos_delivery/waku/waku_store_sync/protocols_metrics.nim index 53595f931..f76db8b47 100644 --- a/logos_delivery/waku/waku_store_sync/protocols_metrics.nim +++ b/logos_delivery/waku/waku_store_sync/protocols_metrics.nim @@ -17,6 +17,12 @@ declarePublicHistogram reconciliation_differences, declarePublicCounter total_bytes_exchanged, "the number of bytes sent and received by the protocols", ["protocol", "direction"] +declarePublicCounter total_transfer_messages_rejected, + "number of received transfer messages dropped by validation" + +declarePublicCounter total_transfer_messages_unverified, + "number of received transfer messages accepted without proof verification (archives do not persist RLN proofs)" + declarePublicCounter total_transfer_messages_exchanged, "the number of messages sent and received by the transfer protocol", ["direction"] diff --git a/logos_delivery/waku/waku_store_sync/transfer.nim b/logos_delivery/waku/waku_store_sync/transfer.nim index 5d20afb18..caa7b92a5 100644 --- a/logos_delivery/waku/waku_store_sync/transfer.nim +++ b/logos_delivery/waku/waku_store_sync/transfer.nim @@ -31,9 +31,14 @@ import logScope: topics = "waku transfer" +type TransferValidator* = proc(msg: WakuMessage): Future[bool] {.gcsafe, raises: [].} + ## Returns false when a received transfer message must be dropped + ## (e.g. failed RLN proof verification). + type SyncTransfer* = ref object of LPProtocol wakuArchive: WakuArchive peerManager: PeerManager + msgValidator: Opt[TransferValidator] # Send IDs to reconciliation protocol for storage idsTx: AsyncQueue[(SyncID, PubsubTopic, ContentTopic)] @@ -169,8 +174,24 @@ proc initProtocolHandler(self: SyncTransfer) = let hash = computeMessageHash(pubsub, msg) + if self.msgValidator.isSome(): + let valid = + try: + await self.msgValidator.get()(msg) + except CancelledError as exc: + raise exc + except CatchableError: + error "transfer message validation error", + remote_peer_id = conn.peerId, error = getCurrentExceptionMsg() + false + + if not valid: + total_transfer_messages_rejected.inc() + warn "transfer message failed validation, dropping", + remote_peer_id = conn.peerId + continue + try: - #TODO verify msg RLN proof... (await self.wakuArchive.syncMessageIngress(hash, pubsub, msg)).isOkOr: error "failed to archive message", error = $error continue @@ -202,6 +223,7 @@ proc new*( idsTx: AsyncQueue[(SyncID, PubsubTopic, ContentTopic)], localWantsRx: AsyncQueue[PeerId], remoteNeedsRx: AsyncQueue[(PeerId, WakuMessageHash)], + msgValidator: Opt[TransferValidator] = Opt.none(TransferValidator), ): T = var transfer = SyncTransfer( peerManager: peerManager, @@ -209,6 +231,7 @@ proc new*( idsTx: idsTx, localWantsRx: localWantsRx, remoteNeedsRx: remoteNeedsRx, + msgValidator: msgValidator, ) transfer.initProtocolHandler() diff --git a/tests/api/test_api_send.nim b/tests/api/test_api_send.nim index 6470b0168..5d852080c 100644 --- a/tests/api/test_api_send.nim +++ b/tests/api/test_api_send.nim @@ -319,6 +319,53 @@ suite "Waku API - Send": (await node.stop()).isOkOr: raiseAssert "Failed to stop node: " & error + asyncTest "Propagation mode: send confirmed without a store node": + ## send-confirmation=propagation: MessageSent fires from the publish path; + ## no store node is connected (same shape as "Send only propagates", which + ## in store mode never fires MessageSent). + var node: LogosDelivery + lockNewGlobalBrokerContext: + node = ( + await LogosDelivery.new( + LogosDeliveryConf( + kernelConf: KernelConf(createApiNodeConf()), + messagingConf: + Opt.some(MessagingClientConf(sendConfirmation: Opt.some("propagation"))), + channelsConf: Opt.some(ReliableChannelManagerConf()), + ) + ) + ).valueOr: + raiseAssert error + (await node.start()).isOkOr: + raiseAssert "Failed to start Waku node: " & error + + await node.waku.node.connectToNodes(@[relayNode1PeerInfo]) + + let eventManager = newSendEventListenerManager(node.waku.brokerCtx) + defer: + await eventManager.teardown() + + let envelope = MessageEnvelope.init( + ContentTopic("/waku/2/default-content/proto"), "test payload" + ) + + let requestId = (await node.messagingClient.send(envelope)).valueOr: + raiseAssert error + + const eventTimeout = 10.seconds + discard await eventManager.waitForEvents(eventTimeout) + + eventManager.validate( + {SendEventOutcome.Sent, SendEventOutcome.Propagated}, requestId + ) + # exactly one confirmation: the finalized task must not be re-reported + # by later service-loop ticks + check eventManager.sentCount == 1 + check eventManager.propagatedCount == 1 + + (await node.stop()).isOkOr: + raiseAssert "Failed to stop node: " & error + asyncTest "Send only propagates fallback to lightpush": var node: LogosDelivery lockNewGlobalBrokerContext: 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/tests/waku_store_sync/test_all.nim b/tests/waku_store_sync/test_all.nim index 82daa388b..bb53db6e5 100644 --- a/tests/waku_store_sync/test_all.nim +++ b/tests/waku_store_sync/test_all.nim @@ -1,3 +1,8 @@ {.used.} -import ./test_protocol, ./test_storage, ./test_codec +import + ./test_protocol, + ./test_storage, + ./test_codec, + ./test_full_node_sync, + ./test_startup_catchup diff --git a/tests/waku_store_sync/test_full_node_sync.nim b/tests/waku_store_sync/test_full_node_sync.nim new file mode 100644 index 000000000..6f1377406 --- /dev/null +++ b/tests/waku_store_sync/test_full_node_sync.nim @@ -0,0 +1,180 @@ +{.used.} + +## Integration test for the "store as a startup-only dependency" experiment: +## two full nodes (no store service) wired exactly like `mountStoreSync` — +## reconciliation + transfer sharing one peer manager and the three async +## queues, backed by small in-memory archives — must recover missed +## messages from each other end to end. + +import results, testutils/unittests, chronos, chronicles +import + ../../logos_delivery/waku/[ + node/peer_manager, + waku_core, + waku_core/message/digest, + waku_store_sync/common, + waku_store_sync/reconciliation, + waku_store_sync/transfer, + waku_archive/archive, + waku_archive/driver, + waku_archive/common, + ], + ../testlib/[wakucore, testasync], + ../waku_archive/archive_utils + +type FullNode = object + switch: Switch + driver: ArchiveDriver + archive: WakuArchive + recon: SyncReconciliation + transfer: SyncTransfer + peerInfo: RemotePeerInfo + peerManager: PeerManager + +proc newFullNode( + msgValidator: Opt[TransferValidator] = Opt.none(TransferValidator) +): Future[FullNode] {.async.} = + ## Mirrors the wiring of WakuNode.mountStoreSync: one peer manager and + ## three shared channels connect reconciliation and transfer. + let switch = newTestSwitch() + await switch.start() + + let driver = newSqliteArchiveDriver() + let archive = newWakuArchive(driver) + let peerManager = PeerManager.new(switch) + + let idsChannel = newAsyncQueue[(SyncID, PubsubTopic, ContentTopic)]() + let wantsChannel = newAsyncQueue[PeerId]() + let needsChannel = newAsyncQueue[(PeerId, WakuMessageHash)]() + + let recon = ( + await SyncReconciliation.new( + pubsubTopics = @[], + contentTopics = @[], + peerManager = peerManager, + wakuArchive = archive, + relayJitter = 0.seconds, + idsRx = idsChannel, + localWantsTx = wantsChannel, + remoteNeedsTx = needsChannel, + ) + ).valueOr: + raiseAssert error + + await recon.start() + switch.mount(recon) + + let transfer = SyncTransfer.new( + peerManager = peerManager, + wakuArchive = archive, + idsTx = idsChannel, + localWantsRx = wantsChannel, + remoteNeedsRx = needsChannel, + msgValidator = msgValidator, + ) + + await transfer.start() + switch.mount(transfer) + + return FullNode( + switch: switch, + driver: driver, + archive: archive, + recon: recon, + transfer: transfer, + peerInfo: switch.peerInfo.toRemotePeerInfo(), + peerManager: peerManager, + ) + +proc stop(node: FullNode) {.async.} = + await node.transfer.stop() + await node.recon.stop() + await node.switch.stop() + +proc insertMessage(node: FullNode, msg: WakuMessage) = + ## Live-ingest path: relay would archive the message and feed the + ## reconciliation storage (subscription_manager archive + sync handlers). + discard node.driver.put(DefaultPubsubTopic, @[msg]) + node.recon.messageIngress(DefaultPubsubTopic, msg) + +proc hasMessage(node: FullNode, hash: WakuMessageHash): Future[bool] {.async.} = + var query = ArchiveQuery() + query.includeData = true + query.hashes = @[hash] + + let response = (await node.archive.findMessages(query)).valueOr: + raiseAssert $error + + return response.messages.len > 0 + +suite "Waku Sync: full node miss recovery": + var nodeA {.threadvar.}: FullNode + var nodeB {.threadvar.}: FullNode + + asyncSetup: + nodeA = await newFullNode() + nodeB = await newFullNode() + + nodeA.peerManager.addPeer(nodeB.peerInfo) + nodeB.peerManager.addPeer(nodeA.peerInfo) + + asyncTeardown: + await nodeA.stop() + await nodeB.stop() + + asyncTest "node recovers a missed message from a full-node peer": + let msg = fakeWakuMessage(contentTopic = DefaultContentTopic) + let hash = computeMessageHash(DefaultPubsubTopic, msg) + + nodeA.insertMessage(msg) + + check not await nodeB.hasMessage(hash) + + let res = await nodeB.recon.storeSynchronization(Opt.some(nodeA.peerInfo)) + assert res.isOk(), $res.error + + # transfer of the missing message happens asynchronously after the + # reconciliation session ends + await sleepAsync(1.seconds) + + check await nodeB.hasMessage(hash) + + asyncTest "one session converges both nodes to the union": + let msgA = fakeWakuMessage(payload = @[byte 1], contentTopic = DefaultContentTopic) + let msgB = fakeWakuMessage(payload = @[byte 2], contentTopic = DefaultContentTopic) + let hashA = computeMessageHash(DefaultPubsubTopic, msgA) + let hashB = computeMessageHash(DefaultPubsubTopic, msgB) + + nodeA.insertMessage(msgA) + nodeB.insertMessage(msgB) + + let res = await nodeB.recon.storeSynchronization(Opt.some(nodeA.peerInfo)) + assert res.isOk(), $res.error + + await sleepAsync(1.seconds) + + check: + await nodeA.hasMessage(hashB) + await nodeB.hasMessage(hashA) + + asyncTest "messages failing transfer validation are not archived": + await nodeB.stop() + + let rejectAll: TransferValidator = proc(msg: WakuMessage): Future[bool] {.async.} = + return false + + nodeB = await newFullNode(msgValidator = Opt.some(rejectAll)) + nodeA.peerManager.addPeer(nodeB.peerInfo) + nodeB.peerManager.addPeer(nodeA.peerInfo) + + let msg = fakeWakuMessage(contentTopic = DefaultContentTopic) + let hash = computeMessageHash(DefaultPubsubTopic, msg) + + nodeA.insertMessage(msg) + + let res = await nodeB.recon.storeSynchronization(Opt.some(nodeA.peerInfo)) + assert res.isOk(), $res.error + + await sleepAsync(1.seconds) + + check not await nodeB.hasMessage(hash) diff --git a/tests/waku_store_sync/test_startup_catchup.nim b/tests/waku_store_sync/test_startup_catchup.nim new file mode 100644 index 000000000..a837f3fbf --- /dev/null +++ b/tests/waku_store_sync/test_startup_catchup.nim @@ -0,0 +1,32 @@ +{.used.} + +import testutils/unittests, chronos +import + ../../logos_delivery/waku/node/waku_node, ../../logos_delivery/waku/waku_core/time + +suite "Startup catch-up decision (store as a startup-only dependency)": + const SyncRange = 1800.seconds # 30 min window + let now = getNowInNanosecondTime() + + test "gap within the sync window uses reconciliation": + let lastOnline = now - 300 * 1_000_000_000'i64 # 5 min ago + + check useSyncCatchUp(true, lastOnline, now, SyncRange) + + test "gap at the window boundary uses reconciliation": + let lastOnline = now - 1800 * 1_000_000_000'i64 + + check useSyncCatchUp(true, lastOnline, now, SyncRange) + + test "gap beyond the sync window falls back to store resume": + let lastOnline = now - 2700 * 1_000_000_000'i64 # 45 min ago + + check not useSyncCatchUp(true, lastOnline, now, SyncRange) + + test "fresh start (no last-online record) falls back to store resume": + check not useSyncCatchUp(true, Timestamp(0), now, SyncRange) + + test "without reconciliation mounted always falls back to store resume": + let lastOnline = now - 300 * 1_000_000_000'i64 + + check not useSyncCatchUp(false, lastOnline, now, SyncRange) diff --git a/tests/wakunode_rest/test_rest_store.nim b/tests/wakunode_rest/test_rest_store.nim index 74847bf9b..f7534be7b 100644 --- a/tests/wakunode_rest/test_rest_store.nim +++ b/tests/wakunode_rest/test_rest_store.nim @@ -673,6 +673,82 @@ procSuite "Waku Rest API - Store v3": $response.contentType == $MIMETYPE_JSON response.data.messages.len == 0 + asyncTest "retrieve messages from an archive-only node (no store protocol)": + ## Store-sync standalone nodes carry an archive without mounting the + ## store protocol; REST self-queries must be served from the archive. + + # Given + let node = testWakuNode() + await node.start() + + var restPort = Port(0) + let restAddress = parseIpAddress("0.0.0.0") + let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet() + restPort = restServer.httpServer.address.port # update with bound port for client use + + installStoreApiHandlers(restServer.router, node) + restServer.start() + + # Archive only, deliberately no mountStore + let driver: ArchiveDriver = QueueDriver.new() + let mountArchiveRes = node.mountArchive(driver) + assert mountArchiveRes.isOk(), mountArchiveRes.error + + require node.wakuStore.isNil() + + let msgList = @[ + fakeWakuMessage(@[byte 0], ts = 0), + fakeWakuMessage(@[byte 1], ts = 1), + fakeWakuMessage(@[byte 2], ts = 2), + fakeWakuMessage(@[byte 3], ts = 3), + fakeWakuMessage(@[byte 4], ts = 4), + ] + for msg in msgList: + require (await driver.put(DefaultPubsubTopic, msg)).isOk() + + let client = newRestHttpClient(initTAddress(restAddress, restPort)) + + # Self-query without a peer address is served from the archive + let response = await client.getStoreMessagesV3( + includeData = "true", pubsubTopic = encodeUrl(DefaultPubsubTopic) + ) + + check: + response.status == 200 + $response.contentType == $MIMETYPE_JSON + response.data.messages.len == 5 + + # Cursor pagination works through the archive fallback + var received = newSeq[WakuMessage]() + var reqCursor = "" + for page in 0 ..< 3: + let resp = await client.getStoreMessagesV3( + includeData = "true", + pubsubTopic = encodeUrl(DefaultPubsubTopic), + cursor = reqCursor, + ascending = "true", + pageSize = "2", + ) + + check: + resp.status == 200 + $resp.contentType == $MIMETYPE_JSON + + for element in resp.data.messages: + if element.message.isSome(): + received.add(element.message.get()) + + if resp.data.paginationCursor.isSome(): + reqCursor = resp.data.paginationCursor.get() + else: + break + + check received == msgList + + await restServer.stop() + await restServer.closeWait() + await node.stop() + asyncTest "correct message fields are returned": # Given let node = testWakuNode() 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())