remove call to `waitFor` from clock.now and make it async (#691)
This commit is contained in:
parent
ca0660b07e
commit
591be9446a
|
@ -1,6 +1,5 @@
|
|||
import pkg/chronos
|
||||
import pkg/stew/endians2
|
||||
import pkg/upraises
|
||||
import pkg/stint
|
||||
|
||||
type
|
||||
|
@ -8,7 +7,7 @@ type
|
|||
SecondsSince1970* = int64
|
||||
Timeout* = object of CatchableError
|
||||
|
||||
method now*(clock: Clock): SecondsSince1970 {.base, upraises: [].} =
|
||||
method now*(clock: Clock): Future[SecondsSince1970] {.base, async.} =
|
||||
raiseAssert "not implemented"
|
||||
|
||||
method waitUntil*(clock: Clock, time: SecondsSince1970) {.base, async.} =
|
||||
|
|
|
@ -46,12 +46,12 @@ method stop*(clock: OnChainClock) {.async.} =
|
|||
await clock.subscription.unsubscribe()
|
||||
clock.started = false
|
||||
|
||||
method now*(clock: OnChainClock): SecondsSince1970 =
|
||||
method now*(clock: OnChainClock): Future[SecondsSince1970] {.async.} =
|
||||
when codex_use_hardhat:
|
||||
# hardhat's latest block.timestamp is usually 1s behind the block timestamp
|
||||
# in the newHeads event. When testing, always return the latest block.
|
||||
try:
|
||||
if queriedBlock =? (waitFor clock.provider.getBlock(BlockTag.latest)):
|
||||
if queriedBlock =? (await clock.provider.getBlock(BlockTag.latest)):
|
||||
trace "using last block timestamp for clock.now",
|
||||
lastBlockTimestamp = queriedBlock.timestamp.truncate(int64),
|
||||
cachedBlockTimestamp = clock.lastBlockTime.truncate(int64)
|
||||
|
@ -65,6 +65,6 @@ method now*(clock: OnChainClock): SecondsSince1970 =
|
|||
return toUnix(getTime() + clock.offset)
|
||||
|
||||
method waitUntil*(clock: OnChainClock, time: SecondsSince1970) {.async.} =
|
||||
while (let difference = time - clock.now(); difference > 0):
|
||||
while (let difference = time - (await clock.now()); difference > 0):
|
||||
clock.newBlock.clear()
|
||||
discard await clock.newBlock.wait().withTimeout(chronos.seconds(difference))
|
||||
|
|
|
@ -53,7 +53,7 @@ proc populate*(purchasing: Purchasing,
|
|||
if result.ask.proofProbability == 0.u256:
|
||||
result.ask.proofProbability = purchasing.proofProbability
|
||||
if result.expiry == 0.u256:
|
||||
result.expiry = (purchasing.clock.now().u256 + purchasing.requestExpiryInterval)
|
||||
result.expiry = (await purchasing.clock.now()).u256 + purchasing.requestExpiryInterval
|
||||
if result.nonce == Nonce.default:
|
||||
var id = result.nonce.toArray
|
||||
doAssert randomBytes(id) == 32
|
||||
|
|
|
@ -328,10 +328,12 @@ proc initPurchasingApi(node: CodexNodeRef, router: var RestRouter) =
|
|||
if node.clock.isNil:
|
||||
return RestApiResponse.error(Http500)
|
||||
|
||||
if expiry <= node.clock.now.u256:
|
||||
let now = (await node.clock.now).u256
|
||||
|
||||
if expiry <= now:
|
||||
return RestApiResponse.error(Http400, "Expiry needs to be in future")
|
||||
|
||||
if expiry > node.clock.now.u256 + params.duration:
|
||||
if expiry > now + params.duration:
|
||||
return RestApiResponse.error(Http400, "Expiry has to be before the request's end (now + duration)")
|
||||
|
||||
without purchaseId =? await node.requestStorage(
|
||||
|
|
|
@ -73,8 +73,9 @@ proc subscribeCancellation(agent: SalesAgent) {.async.} =
|
|||
return
|
||||
|
||||
while true:
|
||||
let deadline = max(clock.now, request.expiry.truncate(int64)) + 1
|
||||
trace "Waiting for request to be cancelled", now=clock.now, expiry=deadline
|
||||
let now = await clock.now
|
||||
let deadline = max(now, request.expiry.truncate(int64)) + 1
|
||||
trace "Waiting for request to be cancelled", now=now, expiry=deadline
|
||||
await clock.waitUntil(deadline)
|
||||
|
||||
without state =? await agent.retrieveRequestState():
|
||||
|
@ -85,7 +86,7 @@ proc subscribeCancellation(agent: SalesAgent) {.async.} =
|
|||
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
|
||||
debug "The request is not yet canceled, even though it should be. Waiting for some more time.", currentState = state, now=now
|
||||
|
||||
data.cancelled = onCancelled()
|
||||
|
||||
|
|
|
@ -57,7 +57,7 @@ proc proveLoop(
|
|||
|
||||
proc getCurrentPeriod(): Future[Period] {.async.} =
|
||||
let periodicity = await market.periodicity()
|
||||
return periodicity.periodOf(clock.now().u256)
|
||||
return periodicity.periodOf((await clock.now()).u256)
|
||||
|
||||
proc waitUntilPeriod(period: Period) {.async.} =
|
||||
let periodicity = await market.periodicity()
|
||||
|
|
|
@ -59,7 +59,7 @@ proc deleteExpiredBlock(self: BlockMaintainer, cid: Cid): Future[void] {.async.}
|
|||
trace "Unable to delete block from repoStore"
|
||||
|
||||
proc processBlockExpiration(self: BlockMaintainer, be: BlockExpiration): Future[void] {.async} =
|
||||
if be.expiration < self.clock.now:
|
||||
if be.expiration < (await self.clock.now):
|
||||
await self.deleteExpiredBlock(be.cid)
|
||||
else:
|
||||
inc self.offset
|
||||
|
|
|
@ -229,12 +229,12 @@ proc getBlockExpirationEntry(
|
|||
proc getBlockExpirationEntry(
|
||||
self: RepoStore,
|
||||
cid: Cid,
|
||||
ttl: ?Duration): ?!BatchEntry =
|
||||
ttl: ?Duration): Future[?!BatchEntry] {.async.} =
|
||||
## Get an expiration entry for a batch for duration since "now"
|
||||
##
|
||||
|
||||
let duration = ttl |? self.blockTtl
|
||||
self.getBlockExpirationEntry(cid, self.clock.now() + duration.seconds)
|
||||
self.getBlockExpirationEntry(cid, (await self.clock.now()) + duration.seconds)
|
||||
|
||||
method ensureExpiry*(
|
||||
self: RepoStore,
|
||||
|
@ -340,7 +340,7 @@ method putBlock*(
|
|||
trace "Updating quota", used
|
||||
batch.add((QuotaUsedKey, @(used.uint64.toBytesBE)))
|
||||
|
||||
without blockExpEntry =? self.getBlockExpirationEntry(blk.cid, ttl), err:
|
||||
without blockExpEntry =? (await self.getBlockExpirationEntry(blk.cid, ttl)), err:
|
||||
trace "Unable to create block expiration metadata key", err = err.msg
|
||||
return failure(err)
|
||||
batch.add(blockExpEntry)
|
||||
|
|
|
@ -1,10 +1,10 @@
|
|||
import std/times
|
||||
import pkg/upraises
|
||||
import pkg/chronos
|
||||
import ./clock
|
||||
|
||||
type
|
||||
SystemClock* = ref object of Clock
|
||||
|
||||
method now*(clock: SystemClock): SecondsSince1970 {.upraises: [].} =
|
||||
method now*(clock: SystemClock): Future[SecondsSince1970] {.async.} =
|
||||
let now = times.now().utc
|
||||
now.toTime().toUnix()
|
||||
|
|
|
@ -34,11 +34,11 @@ proc new*(
|
|||
proc slots*(validation: Validation): seq[SlotId] =
|
||||
validation.slots.toSeq
|
||||
|
||||
proc getCurrentPeriod(validation: Validation): UInt256 =
|
||||
return validation.periodicity.periodOf(validation.clock.now().u256)
|
||||
proc getCurrentPeriod(validation: Validation): Future[UInt256] {.async.} =
|
||||
return validation.periodicity.periodOf((await validation.clock.now()).u256)
|
||||
|
||||
proc waitUntilNextPeriod(validation: Validation) {.async.} =
|
||||
let period = validation.getCurrentPeriod()
|
||||
let period = await validation.getCurrentPeriod()
|
||||
let periodEnd = validation.periodicity.periodEnd(period)
|
||||
trace "Waiting until next period", currentPeriod = period
|
||||
await validation.clock.waitUntil(periodEnd.truncate(int64) + 1)
|
||||
|
@ -66,7 +66,7 @@ proc markProofAsMissing(validation: Validation,
|
|||
slotId: SlotId,
|
||||
period: Period) {.async.} =
|
||||
logScope:
|
||||
currentPeriod = validation.getCurrentPeriod()
|
||||
currentPeriod = (await validation.getCurrentPeriod())
|
||||
|
||||
try:
|
||||
if await validation.market.canProofBeMarkedAsMissing(slotId, period):
|
||||
|
@ -82,7 +82,7 @@ proc markProofAsMissing(validation: Validation,
|
|||
|
||||
proc markProofsAsMissing(validation: Validation) {.async.} =
|
||||
for slotId in validation.slots:
|
||||
let previousPeriod = validation.getCurrentPeriod() - 1
|
||||
let previousPeriod = (await validation.getCurrentPeriod()) - 1
|
||||
await validation.markProofAsMissing(slotId, previousPeriod)
|
||||
|
||||
proc run(validation: Validation) {.async.} =
|
||||
|
|
|
@ -32,11 +32,11 @@ proc set*(clock: MockClock, time: SecondsSince1970) =
|
|||
proc advance*(clock: MockClock, seconds: int64) =
|
||||
clock.set(clock.time + seconds)
|
||||
|
||||
method now*(clock: MockClock): SecondsSince1970 =
|
||||
method now*(clock: MockClock): Future[SecondsSince1970] {.async.} =
|
||||
clock.time
|
||||
|
||||
method waitUntil*(clock: MockClock, time: SecondsSince1970) {.async.} =
|
||||
if time > clock.now():
|
||||
if time > (await clock.now()):
|
||||
let future = newFuture[void]()
|
||||
clock.waiting.add(Waiting(until: time, future: future))
|
||||
await future
|
||||
|
|
|
@ -98,7 +98,7 @@ asyncchecksuite "Test Node - Host contracts":
|
|||
test "onExpiryUpdate callback":
|
||||
let
|
||||
# The blocks have set default TTL, so in order to update it we have to have larger TTL
|
||||
expectedExpiry: SecondsSince1970 = clock.now + DefaultBlockTtl.seconds + 11123
|
||||
expectedExpiry: SecondsSince1970 = (await clock.now) + DefaultBlockTtl.seconds + 11123
|
||||
expiryUpdateCallback = !sales.onExpiryUpdate
|
||||
|
||||
(await expiryUpdateCallback(manifestCidStr, expectedExpiry)).tryGet()
|
||||
|
|
|
@ -67,7 +67,7 @@ asyncchecksuite "Sales - start":
|
|||
sales.onProve = proc(slot: Slot, challenge: ProofChallenge): Future[?!seq[byte]] {.async.} =
|
||||
return success(proof)
|
||||
itemsProcessed = @[]
|
||||
request.expiry = (clock.now() + 42).u256
|
||||
request.expiry = ((await clock.now()) + 42).u256
|
||||
|
||||
teardown:
|
||||
await sales.stop()
|
||||
|
|
|
@ -161,7 +161,7 @@ checksuite "Purchasing state machine":
|
|||
market.requestState[request5.id] = RequestState.Failed
|
||||
|
||||
# ensure the started state doesn't error, giving a false positive test result
|
||||
market.requestEnds[request2.id] = clock.now() - 1
|
||||
market.requestEnds[request2.id] = (await clock.now()) - 1
|
||||
|
||||
await purchasing.load()
|
||||
check eventually purchasing.getPurchase(PurchaseId(request1.id)).?finished == false.some
|
||||
|
@ -182,7 +182,7 @@ checksuite "Purchasing state machine":
|
|||
test "moves to PurchaseStarted when request state is Started":
|
||||
let request = StorageRequest.example
|
||||
let purchase = Purchase.new(request, market, clock)
|
||||
market.requestEnds[request.id] = clock.now() + request.ask.duration.truncate(int64)
|
||||
market.requestEnds[request.id] = (await clock.now()) + request.ask.duration.truncate(int64)
|
||||
market.requested = @[request]
|
||||
market.requestState[request.id] = RequestState.Started
|
||||
let next = await PurchaseUnknown().run(purchase)
|
||||
|
@ -215,7 +215,7 @@ checksuite "Purchasing state machine":
|
|||
test "moves to PurchaseFailed state once RequestFailed emitted":
|
||||
let request = StorageRequest.example
|
||||
let purchase = Purchase.new(request, market, clock)
|
||||
market.requestEnds[request.id] = clock.now() + request.ask.duration.truncate(int64)
|
||||
market.requestEnds[request.id] = (await clock.now()) + request.ask.duration.truncate(int64)
|
||||
let future = PurchaseStarted().run(purchase)
|
||||
|
||||
market.emitRequestFailed(request.id)
|
||||
|
@ -226,7 +226,7 @@ checksuite "Purchasing state machine":
|
|||
test "moves to PurchaseFinished state once request finishes":
|
||||
let request = StorageRequest.example
|
||||
let purchase = Purchase.new(request, market, clock)
|
||||
market.requestEnds[request.id] = clock.now() + request.ask.duration.truncate(int64)
|
||||
market.requestEnds[request.id] = (await clock.now()) + request.ask.duration.truncate(int64)
|
||||
let future = PurchaseStarted().run(purchase)
|
||||
|
||||
clock.advance(request.ask.duration.truncate(int64))
|
||||
|
|
|
@ -1,14 +1,15 @@
|
|||
import std/times
|
||||
import std/unittest
|
||||
|
||||
import codex/systemclock
|
||||
|
||||
import ../asynctest
|
||||
import ./helpers
|
||||
|
||||
checksuite "SystemClock":
|
||||
asyncchecksuite "SystemClock":
|
||||
test "Should get now":
|
||||
let clock = SystemClock.new()
|
||||
|
||||
let expectedNow = times.now().utc
|
||||
let now = clock.now()
|
||||
let now = (await clock.now())
|
||||
|
||||
check now == expectedNow.toTime().toUnix()
|
||||
|
|
|
@ -31,9 +31,9 @@ asyncchecksuite "validation":
|
|||
teardown:
|
||||
await validation.stop()
|
||||
|
||||
proc advanceToNextPeriod =
|
||||
proc advanceToNextPeriod {.async.} =
|
||||
let periodicity = Periodicity(seconds: period.u256)
|
||||
let period = periodicity.periodOf(clock.now().u256)
|
||||
let period = periodicity.periodOf((await clock.now()).u256)
|
||||
let periodEnd = periodicity.periodEnd(period)
|
||||
clock.set((periodEnd + 1).truncate(int))
|
||||
|
||||
|
|
|
@ -17,23 +17,23 @@ ethersuite "On-Chain Clock":
|
|||
test "returns the current time of the EVM":
|
||||
let latestBlock = (!await ethProvider.getBlock(BlockTag.latest))
|
||||
let timestamp = latestBlock.timestamp.truncate(int64)
|
||||
check clock.now() == timestamp
|
||||
check (await clock.now()) == timestamp
|
||||
|
||||
test "updates time with timestamp of new blocks":
|
||||
let future = (getTime() + 42.years).toUnix
|
||||
discard await ethProvider.send("evm_setNextBlockTimestamp", @[%future])
|
||||
discard await ethProvider.send("evm_mine")
|
||||
check clock.now() == future
|
||||
check (await clock.now()) == future
|
||||
|
||||
test "can wait until a certain time is reached by the chain":
|
||||
let future = clock.now() + 42 # seconds
|
||||
let future = (await clock.now()) + 42 # seconds
|
||||
let waiting = clock.waitUntil(future)
|
||||
discard await ethProvider.send("evm_setNextBlockTimestamp", @[%future])
|
||||
discard await ethProvider.send("evm_mine")
|
||||
check await waiting.withTimeout(chronos.milliseconds(100))
|
||||
|
||||
test "can wait until a certain time is reached by the wall-clock":
|
||||
let future = clock.now() + 1 # seconds
|
||||
let future = (await clock.now()) + 1 # seconds
|
||||
let waiting = clock.waitUntil(future)
|
||||
check await waiting.withTimeout(chronos.seconds(2))
|
||||
|
||||
|
|
Loading…
Reference in New Issue