Merge branch 'master' into fix-marketplace-block-expiry

Author: benbierens

codex/erasure/erasure.nim

@@ -419,8 +419,8 @@ proc encodeData(
 
         trace "Adding parity block", cid = blk.cid, idx
         cids[idx] = blk.cid
-        if isErr (await self.store.putBlock(blk)):
-          trace "Unable to store block!", cid = blk.cid
+        if error =? (await self.store.putBlock(blk)).errorOption:
+          warn "Unable to store block!", cid = blk.cid, msg = error.msg
           return failure("Unable to store block!")
         idx.inc(params.steps)
 
@@ -619,8 +619,8 @@ proc decode*(self: Erasure, encoded: Manifest): Future[?!Manifest] {.async.} =
             return failure(error)
 
           trace "Recovered block", cid = blk.cid, index = i
-          if isErr (await self.store.putBlock(blk)):
-            trace "Unable to store block!", cid = blk.cid
+          if error =? (await self.store.putBlock(blk)).errorOption:
+            warn "Unable to store block!", cid = blk.cid, msg = error.msg
             return failure("Unable to store block!")
 
           cids[idx] = blk.cid

codex/stores/maintenance.nim

@@ -23,6 +23,9 @@ import ../clock
 import ../logutils
 import ../systemclock
 
+logScope:
+  topics = "codex maintenance"
+
 const
   DefaultBlockInterval* = 10.minutes
   DefaultNumBlocksPerInterval* = 1000
@@ -40,7 +43,7 @@ proc new*(
     repoStore: RepoStore,
     interval: Duration,
     numberOfBlocksPerInterval = 100,
-    timer = Timer.new(),
+    timer = Timer.new("maintenance"),
     clock: Clock = SystemClock.new(),
 ): BlockMaintainer =
   ## Create new BlockMaintainer instance
@@ -59,8 +62,8 @@ proc new*(
 proc deleteExpiredBlock(
     self: BlockMaintainer, cid: Cid
 ): Future[void] {.async: (raises: [CancelledError]).} =
-  if isErr (await self.repoStore.delBlock(cid)):
-    trace "Unable to delete block from repoStore"
+  if error =? (await self.repoStore.delBlock(cid)).errorOption:
+    warn "Unable to delete block from repoStore", error = error.msg
 
 proc processBlockExpiration(
     self: BlockMaintainer, be: BlockExpiration
@@ -78,13 +81,13 @@ proc runBlockCheck(
   )
 
   without iter =? expirations, err:
-    trace "Unable to obtain blockExpirations iterator from repoStore"
+    warn "Unable to obtain blockExpirations iterator from repoStore", err = err.msg
     return
 
   var numberReceived = 0
   for beFut in iter:
     without be =? (await beFut), err:
-      trace "Unable to obtain blockExpiration from iterator"
+      warn "Unable to obtain blockExpiration from iterator", err = err.msg
       continue
     inc numberReceived
     await self.processBlockExpiration(be)
@@ -94,6 +97,7 @@ proc runBlockCheck(
   # We're at the end of the dataset and should start from 0 next time.
   if numberReceived < self.numberOfBlocksPerInterval:
     self.offset = 0
+    trace "Cycle completed"
 
 proc start*(self: BlockMaintainer) =
   proc onTimer(): Future[void] {.async: (raises: []).} =

codex/utils/timer.nim

@@ -24,7 +24,7 @@ type
     name: string
     loopFuture: Future[void]
 
-proc new*(T: type Timer, timerName = "Unnamed Timer"): Timer =
+proc new*(T: type Timer, timerName: string): Timer =
   ## Create a new Timer intance with the given name
   Timer(name: timerName)
 
@@ -35,6 +35,9 @@ proc timerLoop(timer: Timer) {.async: (raises: []).} =
       await sleepAsync(timer.interval)
   except CancelledError:
     discard # do not propagate as timerLoop is asyncSpawned
+  except CatchableError as err:
+    error "CatchableError in timer loop", name = timer.name, msg = err.msg
+  info "Timer loop has stopped", name = timer.name
 
 method start*(
     timer: Timer, callback: TimerCallback, interval: Duration

tests/codex/utils/testtimer.nim

@@ -36,8 +36,8 @@ asyncchecksuite "Timer":
     timer2.start(lettersCallback, 10.milliseconds)
 
   setup:
-    timer1 = Timer.new()
-    timer2 = Timer.new()
+    timer1 = Timer.new("testtimer1")
+    timer2 = Timer.new("testtimer2")
 
     output = ""
     numbersState = 0