waku-org/nwaku
1
0
Fork 0
mirror of https://github.com/waku-org/nwaku.git synced 2025-02-05 19:43:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

archive: simplify and enhance async retention policy application (#2278)

Browse Source 
* Avoid using timer and just use an infinite async loop that can be
cancelled at any time.
This commit is contained in:
Ivan FB 2023-12-11 08:50:40 +01:00 committed by GitHub
parent 45b0be8e75
commit 77c5ba7669
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 21 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
waku/node/waku_node.nim
View File

@ -740,7 +740,6 @@ proc filterUnsubscribeAll*(node: WakuNode,
# yet incompatible to handle both type of filters - use specific filter registration instead # yet incompatible to handle both type of filters - use specific filter registration instead
## Waku archive ## Waku archive
const WakuArchiveDefaultRetentionPolicyInterval* = 30.minutes
proc mountArchive*(node: WakuNode, proc mountArchive*(node: WakuNode,
driver: ArchiveDriver, driver: ArchiveDriver,
retentionPolicy = none(RetentionPolicy)): retentionPolicy = none(RetentionPolicy)):
@ -760,18 +759,7 @@ proc mountArchive*(node: WakuNode,
except CatchableError: except CatchableError:
return err("exception in mountArchive: " & getCurrentExceptionMsg()) return err("exception in mountArchive: " & getCurrentExceptionMsg())
if retentionPolicy.isSome(): asyncSpawn node.wakuArchive.start()
try:
debug "executing message retention policy"
let retPolRes = waitFor node.wakuArchive.executeMessageRetentionPolicy()
if retPolRes.isErr():
return err("error in mountArchive: " & retPolRes.error)
except CatchableError:
return err("exception in mountArch-ret-pol: " & getCurrentExceptionMsg())
node.wakuArchive.startMessageRetentionPolicyPeriodicTask(
WakuArchiveDefaultRetentionPolicyInterval)
return ok() return ok()
## Waku store ## Waku store
@ -1174,6 +1162,9 @@ proc stop*(node: WakuNode) {.async.} =
if not node.wakuRlnRelay.isNil(): if not node.wakuRlnRelay.isNil():
await node.wakuRlnRelay.stop() await node.wakuRlnRelay.stop()
if not node.wakuArchive.isNil():
await node.wakuArchive.stop()
node.started = false node.started = false
proc isReady*(node: WakuNode): Future[bool] {.async.} = proc isReady*(node: WakuNode): Future[bool] {.async.} =

45
waku/waku_archive/archive.nim
View File

@ -68,6 +68,7 @@ type
driver*: ArchiveDriver # TODO: Make this field private. Remove asterisk driver*: ArchiveDriver # TODO: Make this field private. Remove asterisk
validator: MessageValidator validator: MessageValidator
retentionPolicy: RetentionPolicy retentionPolicy: RetentionPolicy
retPolicyFut: Future[Result[void, string]] ## retention policy cancelable future
proc new*(T: type WakuArchive, proc new*(T: type WakuArchive,
driver: ArchiveDriver, driver: ArchiveDriver,
@ -189,19 +190,25 @@ proc findMessages*(w: WakuArchive, query: ArchiveQuery): Future[ArchiveResult] {
return ok(ArchiveResponse(messages: messages, cursor: cursor)) return ok(ArchiveResponse(messages: messages, cursor: cursor))
# Retention policy # Retention policy
const WakuArchiveDefaultRetentionPolicyInterval* = chronos.minutes(30)
proc loopApplyRetentionPolicy*(w: WakuArchive):
Future[Result[void, string]] {.async.} =
proc executeMessageRetentionPolicy*(w: WakuArchive):
Future[Result[void, string]] {.async.} =
if w.retentionPolicy.isNil(): if w.retentionPolicy.isNil():
return err("retentionPolicy is Nil in executeMessageRetentionPolicy") return err("retentionPolicy is Nil in executeMessageRetentionPolicy")
if w.driver.isNil(): if w.driver.isNil():
return err("driver is Nil in executeMessageRetentionPolicy") return err("driver is Nil in executeMessageRetentionPolicy")
let retPolicyRes = await w.retentionPolicy.execute(w.driver) while true:
if retPolicyRes.isErr(): debug "executing message retention policy"
waku_archive_errors.inc(labelValues = [retPolicyFailure]) let retPolicyRes = await w.retentionPolicy.execute(w.driver)
return err("failed execution of retention policy: " & retPolicyRes.error) if retPolicyRes.isErr():
waku_archive_errors.inc(labelValues = [retPolicyFailure])
error "failed execution of retention policy", error=retPolicyRes.error
await sleepAsync(WakuArchiveDefaultRetentionPolicyInterval)
return ok() return ok()
@ -218,26 +225,8 @@ proc reportStoredMessagesMetric*(w: WakuArchive):
return ok() return ok()
proc startMessageRetentionPolicyPeriodicTask*(w: WakuArchive, proc start*(self: WakuArchive) {.async.} =
interval: timer.Duration) = self.retPolicyFut = self.loopApplyRetentionPolicy()
# Start the periodic message retention policy task
# https://github.com/nim-lang/Nim/issues/17369
var executeRetentionPolicy: CallbackFunc proc stop*(self: WakuArchive) {.async.} =
executeRetentionPolicy = self.retPolicyFut.cancel()
CallbackFunc(
proc (arg: pointer) {.gcsafe, raises: [].} =
try:
let retPolRes = waitFor w.executeMessageRetentionPolicy()
if retPolRes.isErr():
waku_archive_errors.inc(labelValues = [retPolicyFailure])
error "error in periodic retention policy", error = retPolRes.error
except CatchableError:
waku_archive_errors.inc(labelValues = [retPolicyFailure])
error "exception in periodic retention policy",
error = getCurrentExceptionMsg()
discard setTimer(Moment.fromNow(interval), executeRetentionPolicy)
)
discard setTimer(Moment.fromNow(interval), executeRetentionPolicy)