[marketplace] make on chain event callbacks async

codex/contracts/market.nim

@@ -91,7 +91,7 @@ method subscribeRequests(market: OnChainMarket,
                          callback: OnRequest):
                         Future[MarketSubscription] {.async.} =
   proc onEvent(event: StorageRequested) {.upraises:[].} =
-    callback(event.requestId, event.ask)
+    asyncSpawn callback(event.requestId, event.ask)
   let subscription = await market.contract.subscribe(StorageRequested, onEvent)
   return OnChainMarketSubscription(eventSubscription: subscription)
 
@@ -102,7 +102,7 @@ method subscribeSlotFilled*(market: OnChainMarket,
                            Future[MarketSubscription] {.async.} =
   proc onEvent(event: SlotFilled) {.upraises:[].} =
     if event.requestId == requestId and event.slotIndex == slotIndex:
-      callback(event.requestId, event.slotIndex)
+      asyncSpawn callback(event.requestId, event.slotIndex)
   let subscription = await market.contract.subscribe(SlotFilled, onEvent)
   return OnChainMarketSubscription(eventSubscription: subscription)
 
@@ -112,7 +112,7 @@ method subscribeFulfillment(market: OnChainMarket,
                            Future[MarketSubscription] {.async.} =
   proc onEvent(event: RequestFulfilled) {.upraises:[].} =
     if event.requestId == requestId:
-      callback(event.requestId)
+      asyncSpawn callback(event.requestId)
   let subscription = await market.contract.subscribe(RequestFulfilled, onEvent)
   return OnChainMarketSubscription(eventSubscription: subscription)
 
@@ -122,7 +122,7 @@ method subscribeRequestCancelled*(market: OnChainMarket,
                                 Future[MarketSubscription] {.async.} =
   proc onEvent(event: RequestCancelled) {.upraises:[].} =
     if event.requestId == requestId:
-      callback(event.requestId)
+      asyncSpawn callback(event.requestId)
   let subscription = await market.contract.subscribe(RequestCancelled, onEvent)
   return OnChainMarketSubscription(eventSubscription: subscription)
 
@@ -132,7 +132,7 @@ method subscribeRequestFailed*(market: OnChainMarket,
                             Future[MarketSubscription] {.async.} =
   proc onEvent(event: RequestFailed) {.upraises:[].} =
     if event.requestId == requestId:
-      callback(event.requestId)
+      asyncSpawn callback(event.requestId)
   let subscription = await market.contract.subscribe(RequestFailed, onEvent)
   return OnChainMarketSubscription(eventSubscription: subscription)
 

codex/purchasing/states/started.nim

@@ -14,7 +14,7 @@ method enterAsync*(state: PurchaseStarted) {.async.} =
   let request = purchase.request
 
   let failed = newFuture[void]()
-  proc callback(_: RequestId) =
+  proc callback(_: RequestId) {.async.} =
     failed.complete()
   let subscription = await market.subscribeRequestFailed(request.id, callback)
 
@@ -22,8 +22,10 @@ method enterAsync*(state: PurchaseStarted) {.async.} =
   try:
     let fut = await one(ended, failed)
     if fut.id == failed.id:
+      ended.cancel()
       state.switch(PurchaseFailed())
     else:
+      failed.cancel()
       state.switch(PurchaseFinished())
     await subscription.unsubscribe()
   except CatchableError as error:

codex/purchasing/states/submitted.nim

@@ -14,7 +14,7 @@ method enterAsync(state: PurchaseSubmitted) {.async.} =
 
   proc wait {.async.} =
     let done = newFuture[void]()
-    proc callback(_: RequestId) =
+    proc callback(_: RequestId) {.async.} =
       done.complete()
     let request = purchase.request
     let subscription = await market.subscribeFulfillment(request.id, callback)

tests/codex/helpers/mockmarket.nim

@@ -67,7 +67,7 @@ method requestStorage*(market: MockMarket,
   market.requested.add(request)
   var subscriptions = market.subscriptions.onRequest
   for subscription in subscriptions:
-    subscription.callback(request.id, request.ask)
+    await subscription.callback(request.id, request.ask)
   return request
 
 method myRequests*(market: MockMarket): Future[seq[RequestId]] {.async.} =
@@ -106,26 +106,26 @@ proc emitSlotFilled*(market: MockMarket,
   for subscription in subscriptions:
     if subscription.requestId == requestId and
        subscription.slotIndex == slotIndex:
-      subscription.callback(requestId, slotIndex)
+      asyncSpawn subscription.callback(requestId, slotIndex)
 
 proc emitRequestCancelled*(market: MockMarket,
                      requestId: RequestId) =
   var subscriptions = market.subscriptions.onRequestCancelled
   for subscription in subscriptions:
     if subscription.requestId == requestId:
-      subscription.callback(requestId)
+      asyncSpawn subscription.callback(requestId)
 
 proc emitRequestFulfilled*(market: MockMarket, requestId: RequestId) =
   var subscriptions = market.subscriptions.onFulfillment
   for subscription in subscriptions:
     if subscription.requestId == requestId:
-      subscription.callback(requestId)
+      asyncSpawn subscription.callback(requestId)
 
 proc emitRequestFailed*(market: MockMarket, requestId: RequestId) =
   var subscriptions = market.subscriptions.onRequestFailed
   for subscription in subscriptions:
     if subscription.requestId == requestId:
-      subscription.callback(requestId)
+      asyncSpawn subscription.callback(requestId)
 
 proc fillSlot*(market: MockMarket,
                requestId: RequestId,