codex-storage
/
nim-codex
mirror of https://github.com/codex-storage/nim-codex.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

adds validation groups and group index in logs of validator

This commit is contained in:
Marcin Czenko 2024-10-14 01:49:39 +02:00
parent 82886b2b5b
commit 1c9228350a
No known key found for this signature in database
GPG Key ID: 33DEA0C8E30937C0
1 changed files with 31 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
codex/validation.nim
View File

@ -47,7 +47,8 @@ proc getCurrentPeriod(validation: Validation): UInt256 =
proc waitUntilNextPeriod(validation: Validation) {.async.} = proc waitUntilNextPeriod(validation: Validation) {.async.} =
let period = validation.getCurrentPeriod() let period = validation.getCurrentPeriod()
let periodEnd = validation.periodicity.periodEnd(period) let periodEnd = validation.periodicity.periodEnd(period)
trace "Waiting until next period", currentPeriod = period trace "Waiting until next period", currentPeriod = period, groups = validation.config.groups,
groupIndex = validation.config.groupIndex
await validation.clock.waitUntil(periodEnd.truncate(int64) + 1) await validation.clock.waitUntil(periodEnd.truncate(int64) + 1)
func groupIndexForSlotId*(slotId: SlotId, func groupIndexForSlotId*(slotId: SlotId,
@ -71,7 +72,8 @@ proc subscribeSlotFilled(validation: Validation) {.async.} =
proc onSlotFilled(requestId: RequestId, slotIndex: UInt256) = proc onSlotFilled(requestId: RequestId, slotIndex: UInt256) =
let slotId = slotId(requestId, slotIndex) let slotId = slotId(requestId, slotIndex)
if validation.shouldValidateSlot(slotId): if validation.shouldValidateSlot(slotId):
trace "Adding slot", slotId trace "Adding slot", slotId, groups = validation.config.groups,
groupIndex = validation.config.groupIndex
validation.slots.incl(slotId) validation.slots.incl(slotId)
let subscription = await validation.market.subscribeSlotFilled(onSlotFilled) let subscription = await validation.market.subscribeSlotFilled(onSlotFilled)
validation.subscriptions.add(subscription) validation.subscriptions.add(subscription)
@ -82,7 +84,8 @@ proc removeSlotsThatHaveEnded(validation: Validation) {.async.} =
for slotId in slots: for slotId in slots:
let state = await validation.market.slotState(slotId) let state = await validation.market.slotState(slotId)
if state != SlotState.Filled: if state != SlotState.Filled:
trace "Removing slot", slotId trace "Removing slot", slotId, groups = validation.config.groups,
groupIndex = validation.config.groupIndex
ended.incl(slotId) ended.incl(slotId)
validation.slots.excl(ended) validation.slots.excl(ended)
@ -94,11 +97,15 @@ proc markProofAsMissing(validation: Validation,
try: try:
if await validation.market.canProofBeMarkedAsMissing(slotId, period): if await validation.market.canProofBeMarkedAsMissing(slotId, period):
trace "Marking proof as missing", slotId, periodProofMissed = period trace "Marking proof as missing", slotId, periodProofMissed = period,
groups = validation.config.groups,
groupIndex = validation.config.groupIndex
await validation.market.markProofAsMissing(slotId, period) await validation.market.markProofAsMissing(slotId, period)
else: else:
let inDowntime {.used.} = await validation.market.inDowntime(slotId) let inDowntime {.used.} = await validation.market.inDowntime(slotId)
trace "Proof not missing", checkedPeriod = period, inDowntime trace "Proof not missing", checkedPeriod = period, inDowntime,
groups = validation.config.groups,
groupIndex = validation.config.groupIndex
except CancelledError: except CancelledError:
raise raise
except CatchableError as e: except CatchableError as e:
@ -111,36 +118,47 @@ proc markProofsAsMissing(validation: Validation) {.async.} =
await validation.markProofAsMissing(slotId, previousPeriod) await validation.markProofAsMissing(slotId, previousPeriod)
proc run(validation: Validation) {.async.} = proc run(validation: Validation) {.async.} =
trace "Validation started" trace "Validation started", groups = validation.config.groups,
groupIndex = validation.config.groupIndex
try: try:
while true: while true:
await validation.waitUntilNextPeriod() await validation.waitUntilNextPeriod()
await validation.removeSlotsThatHaveEnded() await validation.removeSlotsThatHaveEnded()
await validation.markProofsAsMissing() await validation.markProofsAsMissing()
except CancelledError: except CancelledError:
trace "Validation stopped" trace "Validation stopped", groups = validation.config.groups,
groupIndex = validation.config.groupIndex
discard discard
except CatchableError as e: except CatchableError as e:
error "Validation failed", msg = e.msg error "Validation failed", msg = e.msg, groups = validation.config.groups,
groupIndex = validation.config.groupIndex
proc epochForDurationBackFromNow(validation: Validation, proc epochForDurationBackFromNow(validation: Validation,
duration: times.Duration): SecondsSince1970 = duration: times.Duration): SecondsSince1970 =
return validation.clock.now - duration.inSeconds return validation.clock.now - duration.inSeconds
proc restoreHistoricalState(validation: Validation) {.async} = proc restoreHistoricalState(validation: Validation) {.async} =
trace "Restoring historical state..." trace "Restoring historical state...", groups = validation.config.groups,
groupIndex = validation.config.groupIndex
let startTimeEpoch = validation.epochForDurationBackFromNow(MaxStorageRequestDuration) let startTimeEpoch = validation.epochForDurationBackFromNow(MaxStorageRequestDuration)
let slotFilledEvents = await validation.market.queryPastSlotFilledEvents( let slotFilledEvents = await validation.market.queryPastSlotFilledEvents(
fromTime = startTimeEpoch) fromTime = startTimeEpoch)
trace "Found slot filled events", numberOfSlots = slotFilledEvents.len trace "Found slot filled events", numberOfSlots = slotFilledEvents.len,
groups = validation.config.groups,
groupIndex = validation.config.groupIndex
for event in slotFilledEvents: for event in slotFilledEvents:
let slotId = slotId(event.requestId, event.slotIndex) let slotId = slotId(event.requestId, event.slotIndex)
if validation.shouldValidateSlot(slotId): if validation.shouldValidateSlot(slotId):
trace "Adding slot [historical]", slotId trace "Adding slot [historical]", slotId,
groups = validation.config.groups,
groupIndex = validation.config.groupIndex
validation.slots.incl(slotId) validation.slots.incl(slotId)
trace "Removing slots that have ended..." trace "Removing slots that have ended...", groups = validation.config.groups,
groupIndex = validation.config.groupIndex
await removeSlotsThatHaveEnded(validation) await removeSlotsThatHaveEnded(validation)
trace "Historical state restored", numberOfSlots = validation.slots.len trace "Historical state restored", numberOfSlots = validation.slots.len,
groups = validation.config.groups,
groupIndex = validation.config.groupIndex
proc start*(validation: Validation) {.async.} = proc start*(validation: Validation) {.async.} =
validation.periodicity = await validation.market.periodicity() validation.periodicity = await validation.market.periodicity()