fix(sales): replaces `then` with `asyncSpawn` (#1036)

- ensures `addSlotToQueue` does not raise exceptions as it is now asyncSpawned

codex/contracts/market.nim

@@ -115,7 +115,7 @@ method requestStorage(market: OnChainMarket, request: StorageRequest){.async.} =
     await market.approveFunds(request.price())
     discard await market.contract.requestStorage(request).confirm(1)
 
-method getRequest(market: OnChainMarket,
+method getRequest*(market: OnChainMarket,
                   id: RequestId): Future[?StorageRequest] {.async.} =
   convertEthersError:
     try:

codex/sales.nim

@@ -16,8 +16,8 @@ import ./sales/statemachine
 import ./sales/slotqueue
 import ./sales/states/preparing
 import ./sales/states/unknown
-import ./utils/then
 import ./utils/trackedfutures
+import ./utils/exceptions
 
 ## Sales holds a list of available storage that it may sell.
 ##
@@ -325,7 +325,7 @@ proc onSlotFreed(sales: Sales,
 
   trace "slot freed, adding to queue"
 
-  proc addSlotToQueue() {.async.} =
+  proc addSlotToQueue() {.async: (raises: []).} =
     let context = sales.context
     let market = context.market
     let queue = context.slotQueue
@@ -336,25 +336,22 @@ proc onSlotFreed(sales: Sales,
       trace "no existing request metadata, getting request info from contract"
       # if there's no existing slot for that request, retrieve the request
       # from the contract.
+      try:
         without request =? await market.getRequest(requestId):
           error "unknown request in contract"
           return
 
         found = SlotQueueItem.init(request, slotIndex.truncate(uint16))
+      except CancelledError:
+        discard # do not propagate as addSlotToQueue was asyncSpawned
+      except CatchableError as e:
+        error "failed to get request from contract and add slots to queue",
+          error = e.msgDetail
 
     if err =? queue.push(found).errorOption:
-      raise err
+      error "failed to push slot items to queue", error = err.msgDetail
 
-  addSlotToQueue()
+  asyncSpawn addSlotToQueue().track(sales)
-    .track(sales)
-    .catch(proc(err: ref CatchableError) =
-      if err of SlotQueueItemExistsError:
-        error "Failed to push item to queue becaue it already exists"
-      elif err of QueueNotRunningError:
-        warn "Failed to push item to queue becaue queue is not running"
-      else:
-        warn "Error adding request to SlotQueue", error = err.msg
-    )
 
 proc subscribeRequested(sales: Sales) {.async.} =
   let context = sales.context
@@ -482,7 +479,7 @@ proc subscribeSlotReservationsFull(sales: Sales) {.async.} =
   except CatchableError as e:
     error "Unable to subscribe to slot filled events", msg = e.msg
 
-proc startSlotQueue(sales: Sales) {.async.} =
+proc startSlotQueue(sales: Sales) =
   let slotQueue = sales.context.slotQueue
   let reservations = sales.context.reservations
 
@@ -518,7 +515,7 @@ proc unsubscribe(sales: Sales) {.async.} =
 
 proc start*(sales: Sales) {.async.} =
   await sales.load()
-  await sales.startSlotQueue()
+  sales.startSlotQueue()
   await sales.subscribe()
   sales.running = true
 

tests/codex/helpers/mockmarket.nim

@@ -155,7 +155,7 @@ method myRequests*(market: MockMarket): Future[seq[RequestId]] {.async.} =
 method mySlots*(market: MockMarket): Future[seq[SlotId]] {.async.} =
   return market.activeSlots[market.signer]
 
-method getRequest(market: MockMarket,
+method getRequest*(market: MockMarket,
                   id: RequestId): Future[?StorageRequest] {.async.} =
   for request in market.requested:
     if request.id == id:

tests/codex/sales/testsales.nim

@@ -566,6 +566,7 @@ asyncchecksuite "Sales":
     request.ask.slots = 2
     market.requested = @[request]
     market.requestState[request.id] = RequestState.New
+    market.requestEnds[request.id] = request.expiry.toSecondsSince1970
 
     proc fillSlot(slotIdx: UInt256 = 0.u256) {.async.} =
       let address = await market.getSigner()