mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-19 11:09:26 +00:00
feat: startup offline-gap catch-up (store only at fresh start / long offline)
At startup the node estimates its offline gap from the persisted last-online timestamp: within the sync window it catches up via full-node reconciliation (retrying until a sync peer is found, with the periodic sync loop as safety net); beyond the window, or on fresh start, it falls back to a bounded store resume, waiting for a store peer before consuming retry attempts. The catch-up runs as a background future so node startup no longer blocks up to 90 s on store peers, and shutdown bounds its cancellation wait. StoreResume is also mounted on sync-enabled full nodes so they track last-online. Part of the "Store as a startup-only dependency" experiment (step 10). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
ed060553b0
commit
e57546b66f
@ -279,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:
|
||||
|
||||
@ -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
|
||||
@ -379,6 +381,8 @@ proc mountStoreSync*(
|
||||
|
||||
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,
|
||||
@ -613,6 +617,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.
|
||||
@ -626,7 +690,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()
|
||||
@ -699,6 +765,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()
|
||||
|
||||
|
||||
@ -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.} =
|
||||
|
||||
@ -1,3 +1,8 @@
|
||||
{.used.}
|
||||
|
||||
import ./test_protocol, ./test_storage, ./test_codec, ./test_full_node_sync
|
||||
import
|
||||
./test_protocol,
|
||||
./test_storage,
|
||||
./test_codec,
|
||||
./test_full_node_sync,
|
||||
./test_startup_catchup
|
||||
|
||||
32
tests/waku_store_sync/test_startup_catchup.nim
Normal file
32
tests/waku_store_sync/test_startup_catchup.nim
Normal file
@ -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)
|
||||
Loading…
x
Reference in New Issue
Block a user