import std/sequtils import pkg/asynctest/chronos/unittest import pkg/chronicles import pkg/chronos import pkg/datastore import pkg/questionable import pkg/questionable/results import pkg/codex/logutils import pkg/codex/sales/slotqueue import ../helpers import ../helpers/mockmarket import ../examples suite "Slot queue start/stop": var queue: SlotQueue setup: queue = SlotQueue.new() teardown: await queue.stop() test "starts out not running": check not queue.running test "can call start multiple times, and when already running": asyncSpawn queue.start() asyncSpawn queue.start() check queue.running test "can call stop when alrady stopped": await queue.stop() check not queue.running test "can call stop when running": asyncSpawn queue.start() await queue.stop() check not queue.running test "can call stop multiple times": asyncSpawn queue.start() await queue.stop() await queue.stop() check not queue.running suite "Slot queue workers": var queue: SlotQueue proc onProcessSlot(item: SlotQueueItem, doneProcessing: Future[void]) {.async.} = await sleepAsync(1000.millis) # this is not illustrative of the realistic scenario as the # `doneProcessing` future would be passed to another context before being # completed and therefore is not as simple as making the callback async doneProcessing.complete() setup: let request = StorageRequest.example queue = SlotQueue.new(maxSize = 5, maxWorkers = 3) queue.onProcessSlot = onProcessSlot proc startQueue = asyncSpawn queue.start() teardown: await queue.stop() test "activeWorkers should be 0 when not running": check queue.activeWorkers == 0 test "maxWorkers cannot be 0": expect ValueError: discard SlotQueue.new(maxSize = 1, maxWorkers = 0) test "maxWorkers cannot surpass maxSize": expect ValueError: discard SlotQueue.new(maxSize = 1, maxWorkers = 2) test "does not surpass max workers": startQueue() let item1 = SlotQueueItem.example let item2 = SlotQueueItem.example let item3 = SlotQueueItem.example let item4 = SlotQueueItem.example check queue.push(item1).isOk check queue.push(item2).isOk check queue.push(item3).isOk check queue.push(item4).isOk check eventually queue.activeWorkers == 3 test "discards workers once processing completed": proc processSlot(item: SlotQueueItem, done: Future[void]) {.async.} = await sleepAsync(1.millis) done.complete() queue.onProcessSlot = processSlot startQueue() let item1 = SlotQueueItem.example let item2 = SlotQueueItem.example let item3 = SlotQueueItem.example let item4 = SlotQueueItem.example check queue.push(item1).isOk # finishes after 1.millis check queue.push(item2).isOk # finishes after 1.millis check queue.push(item3).isOk # finishes after 1.millis check queue.push(item4).isOk check eventually queue.activeWorkers == 1 suite "Slot queue": var onProcessSlotCalled = false var onProcessSlotCalledWith: seq[(RequestId, uint16)] var queue: SlotQueue var paused: bool proc newSlotQueue(maxSize, maxWorkers: int, processSlotDelay = 1.millis) = queue = SlotQueue.new(maxWorkers, maxSize.uint16) queue.onProcessSlot = proc(item: SlotQueueItem, done: Future[void]) {.async.} = await sleepAsync(processSlotDelay) trace "processing item", requestId = item.requestId, slotIndex = item.slotIndex onProcessSlotCalled = true onProcessSlotCalledWith.add (item.requestId, item.slotIndex) done.complete() asyncSpawn queue.start() setup: onProcessSlotCalled = false onProcessSlotCalledWith = @[] teardown: paused = false await queue.stop() test "starts out empty": newSlotQueue(maxSize = 2, maxWorkers = 2) check queue.len == 0 check $queue == "[]" test "reports correct size": newSlotQueue(maxSize = 2, maxWorkers = 2) check queue.size == 2 test "correctly compares SlotQueueItems": var requestA = StorageRequest.example requestA.ask.duration = 1.u256 requestA.ask.reward = 1.u256 check requestA.ask.pricePerSlot == 1.u256 requestA.ask.collateral = 100000.u256 requestA.expiry = 1001.u256 var requestB = StorageRequest.example requestB.ask.duration = 100.u256 requestB.ask.reward = 1000.u256 check requestB.ask.pricePerSlot == 100000.u256 requestB.ask.collateral = 1.u256 requestB.expiry = 1000.u256 let itemA = SlotQueueItem.init(requestA, 0) let itemB = SlotQueueItem.init(requestB, 0) check itemB < itemA # B higher priority than A check itemA > itemB test "expands available all possible slot indices on init": let request = StorageRequest.example let items = SlotQueueItem.init(request) check items.len.uint64 == request.ask.slots var checked = 0 for slotIndex in 0'u16..