feat: mount store sync independently of the store service

Promote storeSyncConf from StoreServiceConf to a top-level WakuConf
field so relay/full nodes can run RBSR reconciliation + transfer
without mounting the store service. When no store service is present,
back the sync protocols with a small archive (default sqlite://:memory:,
hidden --store-sync-db-url override) bounded by a time:2*range retention
policy. Drive the ENR Sync capability bit from the new top-level conf so
full nodes discover each other for reconciliation.

Existing CLI flags are unchanged; store-service nodes keep identical
behavior, including --storenode sync-peer slotting (standalone nodes
warn instead, so sync peer selection stays discovery-based and is not
pinned to one store peer).

Part of the "Store as a startup-only dependency" experiment (step 1).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

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,

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():

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]

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)

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

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

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