feat: partial payouts for cancelled requests (#561)
This commit is contained in:
parent
8a7d74e6b2
commit
2fc71cf81b
|
@ -54,6 +54,7 @@ type
|
||||||
Finished
|
Finished
|
||||||
Failed
|
Failed
|
||||||
Paid
|
Paid
|
||||||
|
Cancelled
|
||||||
|
|
||||||
proc `==`*(x, y: Nonce): bool {.borrow.}
|
proc `==`*(x, y: Nonce): bool {.borrow.}
|
||||||
proc `==`*(x, y: RequestId): bool {.borrow.}
|
proc `==`*(x, y: RequestId): bool {.borrow.}
|
||||||
|
|
|
@ -22,4 +22,4 @@ method run*(state: PurchaseCancelled, machine: Machine): Future[?State] {.async.
|
||||||
await purchase.market.withdrawFunds(purchase.requestId)
|
await purchase.market.withdrawFunds(purchase.requestId)
|
||||||
|
|
||||||
let error = newException(Timeout, "Purchase cancelled due to timeout")
|
let error = newException(Timeout, "Purchase cancelled due to timeout")
|
||||||
return some State(PurchaseErrored(error: error))
|
purchase.future.fail(error)
|
||||||
|
|
|
@ -33,7 +33,7 @@ method run*(state: PurchaseSubmitted, machine: Machine): Future[?State] {.async.
|
||||||
await subscription.unsubscribe()
|
await subscription.unsubscribe()
|
||||||
|
|
||||||
proc withTimeout(future: Future[void]) {.async.} =
|
proc withTimeout(future: Future[void]) {.async.} =
|
||||||
let expiry = request.expiry.truncate(int64)
|
let expiry = request.expiry.truncate(int64) + 1
|
||||||
await future.withTimeout(clock, expiry)
|
await future.withTimeout(clock, expiry)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
|
|
@ -54,6 +54,11 @@ proc retrieveRequest*(agent: SalesAgent) {.async.} =
|
||||||
if data.request.isNone:
|
if data.request.isNone:
|
||||||
data.request = await market.getRequest(data.requestId)
|
data.request = await market.getRequest(data.requestId)
|
||||||
|
|
||||||
|
proc retrieveRequestState*(agent: SalesAgent): Future[?RequestState] {.async.} =
|
||||||
|
let data = agent.data
|
||||||
|
let market = agent.context.market
|
||||||
|
return await market.requestState(data.requestId)
|
||||||
|
|
||||||
proc subscribeCancellation(agent: SalesAgent) {.async.} =
|
proc subscribeCancellation(agent: SalesAgent) {.async.} =
|
||||||
let data = agent.data
|
let data = agent.data
|
||||||
let clock = agent.context.clock
|
let clock = agent.context.clock
|
||||||
|
@ -62,8 +67,20 @@ proc subscribeCancellation(agent: SalesAgent) {.async.} =
|
||||||
without request =? data.request:
|
without request =? data.request:
|
||||||
return
|
return
|
||||||
|
|
||||||
await clock.waitUntil(request.expiry.truncate(int64))
|
while true:
|
||||||
|
let deadline = max(clock.now, request.expiry.truncate(int64)) + 1
|
||||||
|
trace "Waiting for request to be cancelled", now=clock.now, expiry=deadline
|
||||||
|
await clock.waitUntil(deadline)
|
||||||
|
|
||||||
|
without state =? await agent.retrieveRequestState():
|
||||||
|
error "Uknown request", requestId = data.requestId
|
||||||
|
return
|
||||||
|
|
||||||
|
if state == RequestState.Cancelled:
|
||||||
agent.schedule(cancelledEvent(request))
|
agent.schedule(cancelledEvent(request))
|
||||||
|
break
|
||||||
|
|
||||||
|
debug "The request is not yet canceled, even though it should be. Waiting for some more time.", currentState = state, now=clock.now
|
||||||
|
|
||||||
data.cancelled = onCancelled()
|
data.cancelled = onCancelled()
|
||||||
|
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
import pkg/chronicles
|
import pkg/chronicles
|
||||||
|
import ../salesagent
|
||||||
import ../statemachine
|
import ../statemachine
|
||||||
import ./errorhandling
|
import ./errorhandling
|
||||||
import ./errored
|
import ./errored
|
||||||
|
@ -8,11 +9,30 @@ logScope:
|
||||||
|
|
||||||
type
|
type
|
||||||
SaleCancelled* = ref object of ErrorHandlingState
|
SaleCancelled* = ref object of ErrorHandlingState
|
||||||
SaleCancelledError* = object of CatchableError
|
|
||||||
SaleTimeoutError* = object of SaleCancelledError
|
|
||||||
|
|
||||||
method `$`*(state: SaleCancelled): string = "SaleCancelled"
|
method `$`*(state: SaleCancelled): string = "SaleCancelled"
|
||||||
|
|
||||||
method run*(state: SaleCancelled, machine: Machine): Future[?State] {.async.} =
|
method run*(state: SaleCancelled, machine: Machine): Future[?State] {.async.} =
|
||||||
let error = newException(SaleTimeoutError, "Sale cancelled due to timeout")
|
let agent = SalesAgent(machine)
|
||||||
return some State(SaleErrored(error: error))
|
let data = agent.data
|
||||||
|
let market = agent.context.market
|
||||||
|
|
||||||
|
without request =? data.request:
|
||||||
|
raiseAssert "no sale request"
|
||||||
|
|
||||||
|
without slotIndex =? data.slotIndex:
|
||||||
|
raiseAssert("no slot index assigned")
|
||||||
|
|
||||||
|
let slot = Slot(request: request, slotIndex: slotIndex)
|
||||||
|
debug "Collecting collateral and partial payout", requestId = $data.requestId, slotIndex
|
||||||
|
await market.freeSlot(slot.id)
|
||||||
|
|
||||||
|
if onClear =? agent.context.onClear and
|
||||||
|
request =? data.request and
|
||||||
|
slotIndex =? data.slotIndex:
|
||||||
|
onClear(request, slotIndex)
|
||||||
|
|
||||||
|
if onCleanUp =? agent.onCleanUp:
|
||||||
|
await onCleanUp()
|
||||||
|
|
||||||
|
warn "Sale cancelled due to timeout", requestId = $data.requestId, slotIndex
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
import pkg/chronicles
|
import pkg/chronicles
|
||||||
|
import ../salesagent
|
||||||
import ../statemachine
|
import ../statemachine
|
||||||
import ./errorhandling
|
import ./errorhandling
|
||||||
import ./errored
|
import ./errored
|
||||||
|
@ -13,5 +14,18 @@ type
|
||||||
method `$`*(state: SaleFailed): string = "SaleFailed"
|
method `$`*(state: SaleFailed): string = "SaleFailed"
|
||||||
|
|
||||||
method run*(state: SaleFailed, machine: Machine): Future[?State] {.async.} =
|
method run*(state: SaleFailed, machine: Machine): Future[?State] {.async.} =
|
||||||
|
let data = SalesAgent(machine).data
|
||||||
|
let market = SalesAgent(machine).context.market
|
||||||
|
|
||||||
|
without request =? data.request:
|
||||||
|
raiseAssert "no sale request"
|
||||||
|
|
||||||
|
without slotIndex =? data.slotIndex:
|
||||||
|
raiseAssert("no slot index assigned")
|
||||||
|
|
||||||
|
let slot = Slot(request: request, slotIndex: slotIndex)
|
||||||
|
debug "Removing slot from mySlots", requestId = $data.requestId, slotIndex
|
||||||
|
await market.freeSlot(slot.id)
|
||||||
|
|
||||||
let error = newException(SaleFailedError, "Sale failed")
|
let error = newException(SaleFailedError, "Sale failed")
|
||||||
return some State(SaleErrored(error: error))
|
return some State(SaleErrored(error: error))
|
||||||
|
|
|
@ -54,3 +54,5 @@ method run*(state: SaleUnknown, machine: Machine): Future[?State] {.async.} =
|
||||||
return some State(SaleFinished())
|
return some State(SaleFinished())
|
||||||
of SlotState.Failed:
|
of SlotState.Failed:
|
||||||
return some State(SaleFailed())
|
return some State(SaleFailed())
|
||||||
|
of SlotState.Cancelled:
|
||||||
|
return some State(SaleCancelled())
|
||||||
|
|
|
@ -453,10 +453,35 @@ asyncchecksuite "Sales":
|
||||||
return success()
|
return success()
|
||||||
createAvailability()
|
createAvailability()
|
||||||
await market.requestStorage(request)
|
await market.requestStorage(request)
|
||||||
clock.set(request.expiry.truncate(int64))
|
|
||||||
|
# If we would not await, then the `clock.set` would run "too fast" as the `subscribeCancellation()`
|
||||||
|
# would otherwise not set the timeout early enough as it uses `clock.now` in the deadline calculation.
|
||||||
|
await sleepAsync(chronos.milliseconds(100))
|
||||||
|
market.requestState[request.id]=RequestState.Cancelled
|
||||||
|
clock.set(request.expiry.truncate(int64)+1)
|
||||||
check eventually (await reservations.all(Availability)).get == @[availability]
|
check eventually (await reservations.all(Availability)).get == @[availability]
|
||||||
check getAvailability().size == origSize
|
check getAvailability().size == origSize
|
||||||
|
|
||||||
|
test "verifies that request is indeed expired from onchain before firing onCancelled":
|
||||||
|
let origSize = availability.size
|
||||||
|
sales.onStore = proc(request: StorageRequest,
|
||||||
|
slot: UInt256,
|
||||||
|
onBatch: BatchProc): Future[?!void] {.async.} =
|
||||||
|
await sleepAsync(chronos.hours(1))
|
||||||
|
return success()
|
||||||
|
createAvailability()
|
||||||
|
await market.requestStorage(request)
|
||||||
|
market.requestState[request.id]=RequestState.New # "On-chain" is the request still ongoing even after local expiration
|
||||||
|
|
||||||
|
# If we would not await, then the `clock.set` would run "too fast" as the `subscribeCancellation()`
|
||||||
|
# would otherwise not set the timeout early enough as it uses `clock.now` in the deadline calculation.
|
||||||
|
await sleepAsync(chronos.milliseconds(100))
|
||||||
|
clock.set(request.expiry.truncate(int64)+1)
|
||||||
|
check getAvailability().size == 0
|
||||||
|
|
||||||
|
market.requestState[request.id]=RequestState.Cancelled # Now "on-chain" is also expired
|
||||||
|
check eventually getAvailability().size == origSize
|
||||||
|
|
||||||
test "loads active slots from market":
|
test "loads active slots from market":
|
||||||
let me = await market.getSigner()
|
let me = await market.getSigner()
|
||||||
|
|
||||||
|
|
|
@ -105,7 +105,7 @@ asyncchecksuite "Purchasing":
|
||||||
let purchase = await purchasing.purchase(request)
|
let purchase = await purchasing.purchase(request)
|
||||||
check eventually market.requested.len > 0
|
check eventually market.requested.len > 0
|
||||||
let request = market.requested[0]
|
let request = market.requested[0]
|
||||||
clock.set(request.expiry.truncate(int64))
|
clock.set(request.expiry.truncate(int64) + 1)
|
||||||
expect PurchaseTimeout:
|
expect PurchaseTimeout:
|
||||||
await purchase.wait()
|
await purchase.wait()
|
||||||
|
|
||||||
|
@ -113,7 +113,7 @@ asyncchecksuite "Purchasing":
|
||||||
let purchase = await purchasing.purchase(request)
|
let purchase = await purchasing.purchase(request)
|
||||||
check eventually market.requested.len > 0
|
check eventually market.requested.len > 0
|
||||||
let request = market.requested[0]
|
let request = market.requested[0]
|
||||||
clock.set(request.expiry.truncate(int64))
|
clock.set(request.expiry.truncate(int64) + 1)
|
||||||
expect PurchaseTimeout:
|
expect PurchaseTimeout:
|
||||||
await purchase.wait()
|
await purchase.wait()
|
||||||
check market.withdrawn == @[request.id]
|
check market.withdrawn == @[request.id]
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
import std/random
|
import std/random
|
||||||
|
import std/strutils
|
||||||
import std/sequtils
|
import std/sequtils
|
||||||
import std/times
|
import std/times
|
||||||
import std/typetraits
|
import std/typetraits
|
||||||
|
@ -6,6 +7,12 @@ import pkg/codex/contracts/requests
|
||||||
import pkg/codex/sales/slotqueue
|
import pkg/codex/sales/slotqueue
|
||||||
import pkg/stint
|
import pkg/stint
|
||||||
|
|
||||||
|
proc exampleString*(length: int): string =
|
||||||
|
let chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
|
||||||
|
result = newString(length) # Create a new empty string with a given length
|
||||||
|
for i in 0..<length:
|
||||||
|
result[i] = chars[rand(chars.len-1)] # Generate a random index and set the string's character
|
||||||
|
|
||||||
proc example*[T: SomeInteger](_: type T): T =
|
proc example*[T: SomeInteger](_: type T): T =
|
||||||
rand(T)
|
rand(T)
|
||||||
|
|
||||||
|
|
|
@ -64,10 +64,11 @@ proc getPurchase*(client: CodexClient, purchaseId: PurchaseId): ?!RestPurchase =
|
||||||
let json = ? parseJson(body).catch
|
let json = ? parseJson(body).catch
|
||||||
RestPurchase.fromJson(json)
|
RestPurchase.fromJson(json)
|
||||||
|
|
||||||
proc getSlots*(client: CodexClient): JsonNode =
|
proc getSlots*(client: CodexClient): ?!seq[Slot] =
|
||||||
let url = client.baseurl & "/sales/slots"
|
let url = client.baseurl & "/sales/slots"
|
||||||
let body = client.http.getContent(url)
|
let body = client.http.getContent(url)
|
||||||
parseJson(body).catch |? nil
|
let json = ? parseJson(body).catch
|
||||||
|
seq[Slot].fromJson(json)
|
||||||
|
|
||||||
proc postAvailability*(
|
proc postAvailability*(
|
||||||
client: CodexClient,
|
client: CodexClient,
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
import std/options
|
import std/options
|
||||||
|
import std/sequtils
|
||||||
from pkg/libp2p import `==`
|
from pkg/libp2p import `==`
|
||||||
import pkg/chronos
|
import pkg/chronos
|
||||||
import pkg/stint
|
import pkg/stint
|
||||||
|
@ -10,6 +11,7 @@ import pkg/codex/utils/stintutils
|
||||||
import ../contracts/time
|
import ../contracts/time
|
||||||
import ../contracts/deployment
|
import ../contracts/deployment
|
||||||
import ../codex/helpers
|
import ../codex/helpers
|
||||||
|
import ../examples
|
||||||
import ./twonodes
|
import ./twonodes
|
||||||
|
|
||||||
|
|
||||||
|
@ -164,3 +166,30 @@ twonodessuite "Integration tests", debug1 = false, debug2 = false:
|
||||||
await provider.advanceTime(duration)
|
await provider.advanceTime(duration)
|
||||||
|
|
||||||
check eventually (await token.balanceOf(account2)) - startBalance == duration*reward
|
check eventually (await token.balanceOf(account2)) - startBalance == duration*reward
|
||||||
|
|
||||||
|
test "expired request partially pays out for stored time":
|
||||||
|
let marketplace = Marketplace.new(Marketplace.address, provider.getSigner())
|
||||||
|
let tokenAddress = await marketplace.token()
|
||||||
|
let token = Erc20Token.new(tokenAddress, provider.getSigner())
|
||||||
|
let reward = 400.u256
|
||||||
|
let duration = 100.u256
|
||||||
|
|
||||||
|
# client 2 makes storage available
|
||||||
|
let startBalanceClient2 = await token.balanceOf(account2)
|
||||||
|
discard client2.postAvailability(size=140000.u256, duration=200.u256, minPrice=300.u256, maxCollateral=300.u256).get
|
||||||
|
|
||||||
|
# client 1 requests storage but requires two nodes to host the content
|
||||||
|
let startBalanceClient1 = await token.balanceOf(account1)
|
||||||
|
let expiry = (await provider.currentTime()) + 10
|
||||||
|
let cid = client1.upload(exampleString(100000)).get
|
||||||
|
let id = client1.requestStorage(cid, duration=duration, reward=reward, proofProbability=3.u256, expiry=expiry, collateral=200.u256, nodes=2).get
|
||||||
|
|
||||||
|
# We have to wait for Client 2 fills the slot, before advancing time.
|
||||||
|
# Until https://github.com/codex-storage/nim-codex/issues/594 is implemented nothing better then
|
||||||
|
# sleeping some seconds is available.
|
||||||
|
await sleepAsync(2.seconds)
|
||||||
|
await provider.advanceTimeTo(expiry+1)
|
||||||
|
check eventually(client1.purchaseStateIs(id, "cancelled"), 20000)
|
||||||
|
|
||||||
|
check eventually ((await token.balanceOf(account2)) - startBalanceClient2) > 0 and ((await token.balanceOf(account2)) - startBalanceClient2) < 10*reward
|
||||||
|
check eventually (startBalanceClient1 - (await token.balanceOf(account1))) == ((await token.balanceOf(account2)) - startBalanceClient2)
|
||||||
|
|
|
@ -1 +1 @@
|
||||||
Subproject commit 1854dfba9991a25532de5f6a53cf50e66afb3c8b
|
Subproject commit 14e453ac3150e6c9ca277e605d5df9389ac7eea7
|
Loading…
Reference in New Issue