feat: reduce store dependencies (experimental)

.dockerignore

@@ -7,3 +7,5 @@
 /nimcache
 librln*
 **/vendor/*
+/nimbledeps
+/build

logos_delivery/api/conf/logos_delivery_conf_json.nim

@@ -18,6 +18,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()
@@ -64,7 +66,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)

logos_delivery/api/conf/messaging_conf.nim

@@ -36,6 +36,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]

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)

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(

logos_delivery/waku/common/rate_limit/setting.nim

@@ -11,6 +11,7 @@ type RateLimitedProtocol* = enum
   LIGHTPUSH
   PEEREXCHG
   FILTER
+  STORESYNC
 
 type ProtocolRateLimitSettings* = Table[RateLimitedProtocol, RateLimitSetting]
 
@@ -24,10 +25,19 @@ let UnlimitedRateLimit*: RateLimitSetting = (0, 0.seconds)
 # all subscribed peers
 let FilterDefaultPerPeerRateLimit*: RateLimitSetting = (30, 1.minutes)
 
+# Store sync sessions are heavier than single queries: each one runs a full
+# range diff and may trigger message transfer. Honest peers initiate at most
+# one session per sync interval (a minute or more), so this leaves generous
+# headroom while bounding what a hostile peer can demand for free.
+let StoreSyncDefaultRateLimit*: RateLimitSetting = (30, 5.minutes)
+
 # For being used under GC-safe condition must use threadvar
 var DefaultProtocolRateLimit* {.threadvar.}: ProtocolRateLimitSettings
-DefaultProtocolRateLimit =
-  {GLOBAL: UnlimitedRateLimit, FILTER: FilterDefaultPerPeerRateLimit}.toTable()
+DefaultProtocolRateLimit = {
+  GLOBAL: UnlimitedRateLimit,
+  FILTER: FilterDefaultPerPeerRateLimit,
+  STORESYNC: StoreSyncDefaultRateLimit,
+}.toTable()
 
 proc isUnlimited*(t: RateLimitSetting): bool {.inline.} =
   return t.volume <= 0 or t.period <= 0.seconds
@@ -54,6 +64,8 @@ proc translate(sProtocol: string): RateLimitedProtocol {.raises: [ValueError].}
     return PEEREXCHG
   of "filter":
     return FILTER
+  of "storesync":
+    return STORESYNC
   else:
     raise newException(ValueError, "Unknown protocol definition: " & sProtocol)
 
@@ -83,7 +95,7 @@ proc parse*(
   ## group4: Unit of period - only h:hour, m:minute, s:second, ms:millisecond allowed
   ## whitespaces are allowed lazily
   const parseRegex =
-    """^\s*((store|storev3|lightpush|px|filter)\s*:)?\s*(\d+)\s*\/\s*(\d+)\s*(s|h|m|ms)\s*$"""
+    """^\s*((storesync|store|storev3|lightpush|px|filter)\s*:)?\s*(\d+)\s*\/\s*(\d+)\s*(s|h|m|ms)\s*$"""
   const regexParseSize = re2(parseRegex)
   for settingStr in settings:
     let aSetting = settingStr.toLower()
@@ -116,6 +128,7 @@ proc parse*(
   # due it is taken for protocols not defined in the list - thus those will not apply accidentally wrong settings.
   discard settingsTable.hasKeyOrPut(GLOBAL, UnlimitedRateLimit)
   discard settingsTable.hasKeyOrPut(FILTER, FilterDefaultPerPeerRateLimit)
+  discard settingsTable.hasKeyOrPut(STORESYNC, StoreSyncDefaultRateLimit)
 
   return ok(settingsTable)
 

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,
       )
     )
   )

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,8 @@ type StoreSyncConfBuilder* = object
   rangeSec*: Opt[uint32]
   intervalSec*: Opt[uint32]
   relayJitterSec*: Opt[uint32]
+  dbUrl*: Opt[string]
+  requireProof*: Opt[bool]
 
 proc init*(T: type StoreSyncConfBuilder): StoreSyncConfBuilder =
   StoreSyncConfBuilder()
@@ -31,6 +35,12 @@ 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 withRequireProof*(b: var StoreSyncConfBuilder, requireProof: bool) =
+  b.requireProof = Opt.some(requireProof)
+
 proc build*(b: StoreSyncConfBuilder): Result[Opt[StoreSyncConf], string] =
   if not b.enabled.get(DefaultStoreSyncEnabled):
     return ok(Opt.none(StoreSyncConf))
@@ -42,12 +52,23 @@ 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,
+        requireProof: b.requireProof.get(false),
       )
     )
   )

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,