fix: sales concurrency bug (#537)

Author: AuHau

codex/sales.nim

@@ -101,11 +101,8 @@ proc remove(sales: Sales, agent: SalesAgent) {.async.} =
   if sales.running:
     sales.agents.keepItIf(it != agent)
 
-proc cleanUp(sales: Sales,
-             agent: SalesAgent,
-             processing: Future[void]) {.async.} =
-  await sales.remove(agent)
-  # signal back to the slot queue to cycle a worker
+proc filled(sales: Sales,
+             processing: Future[void]) =
   if not processing.isNil and not processing.finished():
     processing.complete()
 
@@ -121,7 +118,10 @@ proc processSlot(sales: Sales, item: SlotQueueItem, done: Future[void]) =
   )
 
   agent.context.onCleanUp = proc {.async.} =
-    await sales.cleanUp(agent, done)
+    await sales.remove(agent)
+
+  agent.context.onFilled = some proc(request: StorageRequest, slotIndex: UInt256) =
+      sales.filled(done)
 
   agent.start(SalePreparing())
   sales.agents.add agent

codex/sales/salescontext.nim

@@ -14,6 +14,7 @@ type
     onStore*: ?OnStore
     onClear*: ?OnClear
     onSale*: ?OnSale
+    onFilled*: ?OnFilled
     onCleanUp*: OnCleanUp
     onProve*: ?OnProve
     reservations*: Reservations
@@ -28,4 +29,11 @@ type
                   slotIndex: UInt256) {.gcsafe, upraises: [].}
   OnSale* = proc(request: StorageRequest,
                  slotIndex: UInt256) {.gcsafe, upraises: [].}
+
+  # OnFilled has same function as OnSale, but is kept for internal purposes and should not be set by any external
+  # purposes as it is used for freeing Queue Workers after slot is filled. And the callbacks allows only
+  # one callback to be set, so if some other component would use it, it would override the Slot Queue freeing
+  # mechanism which would lead to blocking of the queue.
+  OnFilled* = proc(request: StorageRequest,
+                 slotIndex: UInt256) {.gcsafe, upraises: [].}
   OnCleanUp* = proc: Future[void] {.gcsafe, upraises: [].}

codex/sales/states/filled.nim

@@ -38,9 +38,11 @@ method run*(state: SaleFilled, machine: Machine): Future[?State] {.async.} =
   if host == me.some:
     info "Slot succesfully filled", requestId = $data.requestId, slotIndex
 
-    if request =? data.request and slotIndex =? data.slotIndex and
-        onSale =? context.onSale:
-      onSale(request, slotIndex)
+    if request =? data.request and slotIndex =? data.slotIndex:
+      if onSale =? context.onSale:
+        onSale(request, slotIndex)
+      if onFilled =? context.onFilled:
+        onFilled(request, slotIndex)
 
     when codex_enable_proof_failures:
       if context.simulateProofFailures > 0:

tests/codex/sales/testsales.nim

@@ -377,7 +377,7 @@ asyncchecksuite "Sales":
     check market.filled[0].proof == proof
     check market.filled[0].host == await market.getSigner()
 
-  test "calls onSale when slot is filled":
+  test "calls onFilled when slot is filled":
     var soldAvailability: Availability
     var soldRequest: StorageRequest
     var soldSlotIndex: UInt256