nim-codex/codex/sales/reservations.nim

350 lines
9.4 KiB
Nim
Raw Normal View History

[marketplace] Add Reservations Module (#340) * [marketplace] reservations module - add de/serialization for Availability - add markUsed/markUnused in persisted availability - add query for unused - add reserve/release - reservation module tests - split ContractInteractions into client contracts and host contracts - remove reservations start/stop as the repo start/stop is being managed by the node - remove dedicated reservations metadata store and use the metadata store from the repo instead - Split ContractInteractions into: - ClientInteractions (with purchasing) - HostInteractions (with sales and proving) - compilation fix for nim 1.2 [repostore] fix started flag, add tests [marketplace] persist slot index For loading the sales state from chain, the slot index was not previously persisted in the contract. Will retrieve the slot index from the contract when the sales state is loaded. * Revert repostore changes In favour of separate PR https://github.com/status-im/nim-codex/pull/374. * remove warnings * clean up * tests: stop repostore during teardown * change constructor type identifier Change Contracts constructor to accept Contracts type instead of ContractInteractions. * change constructor return type to Result instead of Option * fix and split interactions tests * clean up, fix tests * find availability by slot id * remove duplication in host/client interactions * add test for finding availability by slotId * log instead of raiseAssert when failed to mark availability as unused * move to SaleErrored state instead of raiseAssert * remove unneeded reverse It appears that order is not preserved in the repostore, so reversing does not have the intended effect here. * update open api spec for potential rest endpoint errors * move functions about available bytes to repostore * WIP: reserve and release availabilities as needed WIP: not tested yet Availabilities are marked as used when matched (just before downloading starts) so that future matching logic does not match an availability currently in use. As the download progresses, batches of blocks are written to disk, and the equivalent bytes are released from the reservation module. The size of the availability is reduced as well. During a reserve or release operation, availability updates occur after the repo is manipulated. If the availability update operation fails, the reserve or release is rolled back to maintain correct accounting of bytes. Finally, once download completes, or if an error occurs, the availability is marked as unused so future matching can occur. * delete availability when all bytes released * fix tests + cleanup * remove availability from SalesContext callbacks Availability is no longer used past the SaleDownloading state in the state machine. Cleanup of Availability (marking unused) is handled directly in the SaleDownloading state, and no longer in SaleErrored or SaleFinished. Likewise, Availabilities shouldn’t need to be handled on node restart. Additionally, Availability was being passed in SalesContext callbacks, and now that Availability is only used temporarily in the SaleDownloading state, Availability is contextually irrelevant to the callbacks, except in OnStore possibly, though it was not being consumed. * test clean up * - remove availability from callbacks and constructors from previous commit that needed to be removed (oopsie) - fix integration test that checks availabilities - there was a bug fixed that crashed the node due to a missing `return success` in onStore - the test was fixed by ensuring that availabilities are remaining on the node, and the size has been reduced - change Availability back to non-ref object and constructor back to init - add trace logging of all state transitions in state machine - add generally useful trace logging * fixes after rebase 1. Fix onProve callbacks 2. Use Slot type instead of tuple for retrieving active slot. 3. Bump codex-contracts-eth that exposes getActivceSlot call. * swap contracts branch to not support slot collateral Slot collateral changes in the contracts require further changes in the client code, so we’ll skip those changes for now and add in a separate commit. * modify Interactions and Deployment constructors - `HostInteractions` and `ClientInteractions` constructors were simplified to take a contract address and no overloads - `Interactions` prepared simplified so there are no overloads - `Deployment` constructor updated so that it takes an optional string parameter, instead `Option[string]` * Move `batchProc` declaration `batchProc` needs to be consumed by both `node` and `salescontext`, and they can’t reference each other as it creates a circular dependency. * [reservations] rename `available` to `hasAvailable` * [reservations] default error message to inner error msg * add SaleIngored state When a storage request is handled but the request does match availabilities, the sales agent machine is sent to the SaleIgnored state. In addition, the agent is constructed in a way that if the request is ignored, the sales agent is removed from the list of active agents being tracked in the sales module.
2023-04-04 07:05:16 +00:00
## Nim-Codex
## Copyright (c) 2022 Status Research & Development GmbH
## Licensed under either of
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.
import std/typetraits
import pkg/chronos
import pkg/chronicles
import pkg/upraises
import pkg/json_serialization
import pkg/json_serialization/std/options
import pkg/stint
import pkg/stew/byteutils
import pkg/nimcrypto
import pkg/questionable
import pkg/questionable/results
push: {.upraises: [].}
import pkg/datastore
import ../stores
import ../contracts/requests
export requests
logScope:
topics = "reservations"
type
AvailabilityId* = distinct array[32, byte]
Availability* = object
id*: AvailabilityId
size*: UInt256
duration*: UInt256
minPrice*: UInt256
used*: bool
Reservations* = ref object
repo: RepoStore
GetNext* = proc(): Future[?Availability] {.upraises: [], gcsafe, closure.}
AvailabilityIter* = ref object
finished*: bool
next*: GetNext
AvailabilityError* = object of CodexError
AvailabilityAlreadyExistsError* = object of AvailabilityError
AvailabilityReserveFailedError* = object of AvailabilityError
AvailabilityReleaseFailedError* = object of AvailabilityError
AvailabilityDeleteFailedError* = object of AvailabilityError
AvailabilityGetFailedError* = object of AvailabilityError
AvailabilityUpdateFailedError* = object of AvailabilityError
const
SalesKey = (CodexMetaKey / "sales").tryGet # TODO: move to sales module
ReservationsKey = (SalesKey / "reservations").tryGet
proc new*(
T: type Reservations,
repo: RepoStore): Reservations =
T(repo: repo)
proc init*(
_: type Availability,
size: UInt256,
duration: UInt256,
minPrice: UInt256): Availability =
var id: array[32, byte]
doAssert randomBytes(id) == 32
Availability(id: AvailabilityId(id), size: size, duration: duration, minPrice: minPrice)
func toArray*(id: AvailabilityId): array[32, byte] =
array[32, byte](id)
proc `==`*(x, y: AvailabilityId): bool {.borrow.}
proc `==`*(x, y: Availability): bool =
x.id == y.id and
x.size == y.size and
x.duration == y.duration and
x.minPrice == y.minPrice
proc `$`*(id: AvailabilityId): string = id.toArray.toHex
proc toErr[E1: ref CatchableError, E2: AvailabilityError](
e1: E1,
_: type E2,
msg: string = e1.msg): ref E2 =
return newException(E2, msg, e1)
proc writeValue*(
writer: var JsonWriter,
value: SlotId | AvailabilityId) {.upraises:[IOError].} =
mixin writeValue
writer.writeValue value.toArray
proc readValue*[T: SlotId | AvailabilityId](
reader: var JsonReader,
value: var T) {.upraises: [SerializationError, IOError].} =
mixin readValue
value = T reader.readValue(T.distinctBase)
func key(id: AvailabilityId): ?!Key =
(ReservationsKey / id.toArray.toHex)
func key*(availability: Availability): ?!Key =
return availability.id.key
func available*(self: Reservations): uint = self.repo.available
func hasAvailable*(self: Reservations, bytes: uint): bool =
self.repo.available(bytes)
proc exists*(
self: Reservations,
id: AvailabilityId): Future[?!bool] {.async.} =
without key =? id.key, err:
return failure(err)
let exists = await self.repo.metaDs.contains(key)
return success(exists)
proc get*(
self: Reservations,
id: AvailabilityId): Future[?!Availability] {.async.} =
if exists =? (await self.exists(id)) and not exists:
let err = newException(AvailabilityGetFailedError,
"Availability does not exist")
return failure(err)
without key =? id.key, err:
return failure(err.toErr(AvailabilityGetFailedError))
without serialized =? await self.repo.metaDs.get(key), err:
return failure(err.toErr(AvailabilityGetFailedError))
without availability =? Json.decode(serialized, Availability).catch, err:
return failure(err.toErr(AvailabilityGetFailedError))
return success availability
proc update(
self: Reservations,
availability: Availability): Future[?!void] {.async.} =
trace "updating availability", id = availability.id, size = availability.size,
used = availability.used
without key =? availability.key, err:
return failure(err)
if err =? (await self.repo.metaDs.put(
key,
@(availability.toJson.toBytes))).errorOption:
return failure(err.toErr(AvailabilityUpdateFailedError))
return success()
proc delete(
self: Reservations,
id: AvailabilityId): Future[?!void] {.async.} =
trace "deleting availability", id
without availability =? (await self.get(id)), err:
return failure(err)
without key =? availability.key, err:
return failure(err)
if err =? (await self.repo.metaDs.delete(key)).errorOption:
return failure(err.toErr(AvailabilityDeleteFailedError))
return success()
proc reserve*(
self: Reservations,
availability: Availability): Future[?!void] {.async.} =
if exists =? (await self.exists(availability.id)) and exists:
let err = newException(AvailabilityAlreadyExistsError,
"Availability already exists")
return failure(err)
without key =? availability.key, err:
return failure(err)
let bytes = availability.size.truncate(uint)
if reserveErr =? (await self.repo.reserve(bytes)).errorOption:
return failure(reserveErr.toErr(AvailabilityReserveFailedError))
if updateErr =? (await self.update(availability)).errorOption:
# rollback the reserve
trace "rolling back reserve"
if rollbackErr =? (await self.repo.release(bytes)).errorOption:
rollbackErr.parent = updateErr
return failure(rollbackErr)
return failure(updateErr)
return success()
proc release*(
self: Reservations,
id: AvailabilityId,
bytes: uint): Future[?!void] {.async.} =
trace "releasing bytes and updating availability", bytes, id
without var availability =? (await self.get(id)), err:
return failure(err)
without key =? id.key, err:
return failure(err)
if releaseErr =? (await self.repo.release(bytes)).errorOption:
return failure(releaseErr.toErr(AvailabilityReleaseFailedError))
availability.size = (availability.size.truncate(uint) - bytes).u256
template rollbackRelease(e: ref CatchableError) =
trace "rolling back release"
if rollbackErr =? (await self.repo.reserve(bytes)).errorOption:
rollbackErr.parent = e
return failure(rollbackErr)
# remove completely used availabilities
if availability.size == 0.u256:
if err =? (await self.delete(availability.id)).errorOption:
rollbackRelease(err)
return failure(err)
return success()
# persist partially used availability with updated size
if err =? (await self.update(availability)).errorOption:
rollbackRelease(err)
return failure(err)
return success()
proc markUsed*(
self: Reservations,
id: AvailabilityId): Future[?!void] {.async.} =
without var availability =? (await self.get(id)), err:
return failure(err)
availability.used = true
let r = await self.update(availability)
if r.isOk:
trace "availability marked used", id = id.toArray.toHex
return r
proc markUnused*(
self: Reservations,
id: AvailabilityId): Future[?!void] {.async.} =
without var availability =? (await self.get(id)), err:
return failure(err)
availability.used = false
let r = await self.update(availability)
if r.isOk:
trace "availability marked unused", id = id.toArray.toHex
return r
iterator items*(self: AvailabilityIter): Future[?Availability] =
while not self.finished:
yield self.next()
proc availabilities*(
self: Reservations): Future[?!AvailabilityIter] {.async.} =
var iter = AvailabilityIter()
let query = Query.init(ReservationsKey)
without results =? await self.repo.metaDs.query(query), err:
return failure(err)
proc next(): Future[?Availability] {.async.} =
await idleAsync()
iter.finished = results.finished
if not results.finished and
r =? (await results.next()) and
serialized =? r.data and
serialized.len > 0:
return some Json.decode(string.fromBytes(serialized), Availability)
return none Availability
iter.next = next
return success iter
proc unused*(r: Reservations): Future[?!seq[Availability]] {.async.} =
var ret: seq[Availability] = @[]
without availabilities =? (await r.availabilities), err:
return failure(err)
for a in availabilities:
if availability =? (await a) and not availability.used:
ret.add availability
return success(ret)
proc find*(
self: Reservations,
size, duration, minPrice: UInt256,
used: bool): Future[?Availability] {.async.} =
without availabilities =? (await self.availabilities), err:
error "failed to get all availabilities", error = err.msg
return none Availability
for a in availabilities:
if availability =? (await a):
if used == availability.used and
size <= availability.size and
duration <= availability.duration and
minPrice >= availability.minPrice:
trace "availability matched",
used, availUsed = availability.used,
size, availsize = availability.size,
duration, availDuration = availability.duration,
minPrice, availMinPrice = availability.minPrice
return some availability
trace "availiability did not match",
used, availUsed = availability.used,
size, availsize = availability.size,
duration, availDuration = availability.duration,
minPrice, availMinPrice = availability.minPrice