fix(sales): fix marketplace block expiry

codex/node.nim

@@ -623,6 +623,7 @@ proc requestStorage*(
 proc onStore(
     self: CodexNodeRef,
     request: StorageRequest,
+    expiry: SecondsSince1970,
     slotIdx: uint64,
     blocksCb: BlocksCb,
     isRepairing: bool = false,
@@ -651,8 +652,6 @@ proc onStore(
     trace "Unable to create slots builder", err = err.msg
     return failure(err)
 
-  let expiry = request.expiry
-
   if slotIdx > manifest.slotRoots.high.uint64:
     trace "Slot index not in manifest", slotIdx
     return failure(newException(CodexError, "Slot index not in manifest"))
@@ -663,7 +662,7 @@ proc onStore(
     trace "Updating expiry for blocks", blocks = blocks.len
 
     let ensureExpiryFutures =
-      blocks.mapIt(self.networkStore.ensureExpiry(it.cid, expiry.toSecondsSince1970))
+      blocks.mapIt(self.networkStore.ensureExpiry(it.cid, expiry))
 
     let res = await allFinishedFailed[?!void](ensureExpiryFutures)
     if res.failure.len > 0:
@@ -789,11 +788,12 @@ proc start*(self: CodexNodeRef) {.async.} =
   if hostContracts =? self.contracts.host:
     hostContracts.sales.onStore = proc(
         request: StorageRequest,
+        expiry: SecondsSince1970,
         slot: uint64,
         onBatch: BatchProc,
         isRepairing: bool = false,
     ): Future[?!void] {.async: (raw: true, raises: [CancelledError]).} =
-      self.onStore(request, slot, onBatch, isRepairing)
+      self.onStore(request, expiry, slot, onBatch, isRepairing)
 
     hostContracts.sales.onExpiryUpdate = proc(
         rootCid: Cid, expiry: SecondsSince1970

codex/sales.nim

@@ -148,26 +148,12 @@ proc cleanUp(
 
   # Re-add items back into the queue to prevent small availabilities from
   # draining the queue. Seen items will be ordered last.
-  if data.slotIndex <= uint16.high.uint64 and reprocessSlot and request =? data.request:
-    let res =
-      await noCancel sales.context.market.slotCollateral(data.requestId, data.slotIndex)
-    if res.isErr:
-      error "Failed to re-add item back to the slot queue: unable to calculate collateral",
-        error = res.error.msg
-    else:
-      let collateral = res.get()
-      let queue = sales.context.slotQueue
-      var seenItem = SlotQueueItem.init(
-        data.requestId,
-        data.slotIndex.uint16,
-        data.ask,
-        request.expiry,
-        seen = true,
-        collateral = collateral,
-      )
-      trace "pushing ignored item to queue, marked as seen"
-      if err =? queue.push(seenItem).errorOption:
-        error "failed to readd slot to queue", errorType = $(type err), error = err.msg
+  if reprocessSlot and request =? data.request and var item =? agent.data.slotQueueItem:
+    let queue = sales.context.slotQueue
+    item.seen = true
+    trace "pushing ignored item to queue, marked as seen"
+    if err =? queue.push(item).errorOption:
+      error "failed to readd slot to queue", errorType = $(type err), error = err.msg
 
   let fut = sales.remove(agent)
   sales.trackedFutures.track(fut)
@@ -181,8 +167,9 @@ proc processSlot(
 ) {.async: (raises: [CancelledError]).} =
   debug "Processing slot from queue", requestId = item.requestId, slot = item.slotIndex
 
-  let agent =
-    newSalesAgent(sales.context, item.requestId, item.slotIndex, none StorageRequest)
+  let agent = newSalesAgent(
+    sales.context, item.requestId, item.slotIndex, none StorageRequest, some item
+  )
 
   let completed = newAsyncEvent()
 

codex/sales/salesagent.nim

@@ -11,6 +11,7 @@ import ./statemachine
 import ./salescontext
 import ./salesdata
 import ./reservations
+import ./slotqueue
 
 export reservations
 
@@ -42,10 +43,16 @@ proc newSalesAgent*(
     requestId: RequestId,
     slotIndex: uint64,
     request: ?StorageRequest,
+    slotQueueItem = SlotQueueItem.none,
 ): SalesAgent =
   var agent = SalesAgent.new()
   agent.context = context
-  agent.data = SalesData(requestId: requestId, slotIndex: slotIndex, request: request)
+  agent.data = SalesData(
+    requestId: requestId,
+    slotIndex: slotIndex,
+    request: request,
+    slotQueueItem: slotQueueItem,
+  )
   return agent
 
 proc retrieveRequest*(agent: SalesAgent) {.async.} =

codex/sales/salescontext.nim

@@ -28,7 +28,11 @@ type
     gcsafe, async: (raises: [CancelledError])
   .}
   OnStore* = proc(
-    request: StorageRequest, slot: uint64, blocksCb: BlocksCb, isRepairing: bool
+    request: StorageRequest,
+    expiry: SecondsSince1970,
+    slot: uint64,
+    blocksCb: BlocksCb,
+    isRepairing: bool,
   ): Future[?!void] {.gcsafe, async: (raises: [CancelledError]).}
   OnProve* = proc(slot: Slot, challenge: ProofChallenge): Future[?!Groth16Proof] {.
     gcsafe, async: (raises: [CancelledError])

codex/sales/salesdata.nim

@@ -2,6 +2,7 @@ import pkg/chronos
 import ../contracts/requests
 import ../market
 import ./reservations
+import ./slotqueue
 
 type SalesData* = ref object
   requestId*: RequestId
@@ -10,3 +11,4 @@ type SalesData* = ref object
   slotIndex*: uint64
   cancelled*: Future[void]
   reservation*: ?Reservation
+  slotQueueItem*: ?SlotQueueItem

codex/sales/slotqueue.nim

@@ -30,7 +30,7 @@ type
     duration: uint64
     pricePerBytePerSecond: UInt256
     collateral: UInt256 # Collateral computed
-    expiry: uint64
+    expiry: ?uint64
     seen: bool
 
   # don't need to -1 to prevent overflow when adding 1 (to always allow push)
@@ -89,8 +89,9 @@ proc `<`*(a, b: SlotQueueItem): bool =
   scoreA.addIf(a.collateral < b.collateral, 2)
   scoreB.addIf(a.collateral > b.collateral, 2)
 
-  scoreA.addIf(a.expiry > b.expiry, 1)
-  scoreB.addIf(a.expiry < b.expiry, 1)
+  if expiryA =? a.expiry and expiryB =? b.expiry:
+    scoreA.addIf(expiryA > expiryB, 1)
+    scoreB.addIf(expiryA < expiryB, 1)
 
   return scoreA > scoreB
 
@@ -124,7 +125,7 @@ proc init*(
     requestId: RequestId,
     slotIndex: uint16,
     ask: StorageAsk,
-    expiry: uint64,
+    expiry: ?uint64,
     collateral: UInt256,
     seen = false,
 ): SlotQueueItem =
@@ -139,6 +140,17 @@ proc init*(
     seen: seen,
   )
 
+proc init*(
+    _: type SlotQueueItem,
+    requestId: RequestId,
+    slotIndex: uint16,
+    ask: StorageAsk,
+    expiry: uint64,
+    collateral: UInt256,
+    seen = false,
+): SlotQueueItem =
+  SlotQueueItem.init(requestId, slotIndex, ask, some expiry, collateral, seen)
+
 proc init*(
     _: type SlotQueueItem,
     request: StorageRequest,
@@ -151,7 +163,7 @@ proc init*(
     _: type SlotQueueItem,
     requestId: RequestId,
     ask: StorageAsk,
-    expiry: uint64,
+    expiry: ?uint64,
     collateral: UInt256,
 ): seq[SlotQueueItem] {.raises: [SlotsOutOfRangeError].} =
   if not ask.slots.inRange:
@@ -167,10 +179,19 @@ proc init*(
   Rng.instance.shuffle(items)
   return items
 
+proc init*(
+    _: type SlotQueueItem,
+    requestId: RequestId,
+    ask: StorageAsk,
+    expiry: uint64,
+    collateral: UInt256,
+): seq[SlotQueueItem] {.raises: [SlotsOutOfRangeError].} =
+  SlotQueueItem.init(requestId, ask, some expiry, collateral)
+
 proc init*(
     _: type SlotQueueItem, request: StorageRequest, collateral: UInt256
 ): seq[SlotQueueItem] =
-  return SlotQueueItem.init(request.id, request.ask, request.expiry, collateral)
+  return SlotQueueItem.init(request.id, request.ask, uint64.none, collateral)
 
 proc inRange*(val: SomeUnsignedInt): bool =
   val.uint16 in SlotQueueSize.low .. SlotQueueSize.high
@@ -196,6 +217,9 @@ proc collateralPerByte*(self: SlotQueueItem): UInt256 =
 proc seen*(self: SlotQueueItem): bool =
   self.seen
 
+proc `seen=`*(self: var SlotQueueItem, seen: bool) =
+  self.seen = seen
+
 proc running*(self: SlotQueue): bool =
   self.running
 

codex/sales/states/downloading.nim

@@ -38,6 +38,7 @@
   let agent = SalesAgent(machine)
   let data = agent.data
   let context = agent.context
+  let market = context.market
   let reservations = context.reservations
 
   without onStore =? context.onStore:
@@ -69,11 +70,21 @@
     return await reservations.release(reservation.id, reservation.availabilityId, bytes)
 
   try:
-    let slotId = slotId(request.id, data.slotIndex)
-    let isRepairing = (await context.market.slotState(slotId)) == SlotState.Repair
+    let requestId = request.id
+    let slotId = slotId(requestId, data.slotIndex)
+    let requestState = await market.requestState(requestId)
+    let isRepairing = (await market.slotState(slotId)) == SlotState.Repair
+
+    trace "Retrieving expiry"
+    var expiry: SecondsSince1970
+    if state =? requestState and state == RequestState.Started:
+      expiry = await market.getRequestEnd(requestId)
+    else:
+      expiry = await market.requestExpiresAt(requestId)
 
     trace "Starting download"
-    if err =? (await onStore(request, data.slotIndex, onBlocks, isRepairing)).errorOption:
+    if err =?
+        (await onStore(request, expiry, data.slotIndex, onBlocks, isRepairing)).errorOption:
       return some State(SaleErrored(error: err, reprocessSlot: false))
 
     trace "Download complete"

tests/codex/helpers/mockmarket.nim

@@ -14,6 +14,7 @@
 import codex/clock
 
 import ../examples
+import ./mockclock
 
 export market
 export tables
@@ -51,7 +52,7 @@
     errorOnFillSlot*: ?(ref MarketError)
     errorOnFreeSlot*: ?(ref MarketError)
     errorOnGetHost*: ?(ref MarketError)
-    clock: ?Clock
+    clock: Clock
 
   Fulfillment* = object
     requestId*: RequestId
@@ -63,7 +64,7 @@
     host*: Address
     slotIndex*: uint64
     proof*: Groth16Proof
-    timestamp: ?SecondsSince1970
+    timestamp: SecondsSince1970
     collateral*: UInt256
 
   Subscriptions = object
@@ -119,7 +120,7 @@
 proc hash*(requestId: RequestId): Hash =
   hash(requestId.toArray)
 
-proc new*(_: type MockMarket, clock: ?Clock = Clock.none): MockMarket =
+proc new*(_: type MockMarket, clock: Clock = MockClock.new()): MockMarket =
   ## Create a new mocked Market instance
   ##
   let config = MarketplaceConfig(
@@ -181,10 +182,15 @@
 method requestStorage*(
     market: MockMarket, request: StorageRequest
 ) {.async: (raises: [CancelledError, MarketError]).} =
+  let now = market.clock.now()
+  let requestExpiresAt = now + request.expiry.toSecondsSince1970
+  let requestEndsAt = now + request.ask.duration.toSecondsSince1970
   market.requested.add(request)
+  market.requestExpiry[request.id] = requestExpiresAt
+  market.requestEnds[request.id] = requestEndsAt
   var subscriptions = market.subscriptions.onRequest
   for subscription in subscriptions:
-    subscription.callback(request.id, request.ask, request.expiry)
+    subscription.callback(request.id, request.ask, requestExpiresAt.uint64)
 
 method myRequests*(market: MockMarket): Future[seq[RequestId]] {.async.} =
   return market.activeRequests[market.signer]
@@ -308,7 +314,7 @@
     slotIndex: slotIndex,
     proof: proof,
     host: host,
-    timestamp: market.clock .? now,
+    timestamp: market.clock.now,
     collateral: collateral,
   )
   market.filled.add(slot)
@@ -541,15 +547,23 @@
 ): Future[seq[StorageRequested]] {.async.} =
   return market.requested.map(
     request =>
-      StorageRequested(requestId: request.id, ask: request.ask, expiry: request.expiry)
+      StorageRequested(
+        requestId: request.id,
+        ask: request.ask,
+        expiry: market.requestExpiry[request.id].uint64,
+      )
   )
 
 method queryPastStorageRequestedEvents*(
     market: MockMarket, blocksAgo: int
 ): Future[seq[StorageRequested]] {.async.} =
   return market.requested.map(
     request =>
-      StorageRequested(requestId: request.id, ask: request.ask, expiry: request.expiry)
+      StorageRequested(
+        requestId: request.id,
+        ask: request.ask,
+        expiry: market.requestExpiry[request.id].uint64,
+      )
   )
 
 method queryPastSlotFilledEvents*(
@@ -571,10 +585,7 @@
 ): Future[seq[SlotFilled]] {.async.} =
   let filtered = market.filled.filter(
     proc(slot: MockSlot): bool =
-      if timestamp =? slot.timestamp:
-        return timestamp >= fromTime
-      else:
-        true
+      return slot.timestamp >= fromTime
   )
   return filtered.map(
     slot => SlotFilled(requestId: slot.requestId, slotIndex: slot.slotIndex)

tests/codex/node/testcontracts.nim

@@ -116,8 +116,7 @@ asyncchecksuite "Test Node - Host contracts":
     let onStore = !sales.onStore
     var request = StorageRequest.example
     request.content.cid = verifiableBlock.cid
-    request.expiry =
-      (getTime() + DefaultBlockTtl.toTimesDuration + 1.hours).toUnix.uint64
+    let expiry = (getTime() + DefaultBlockTtl.toTimesDuration + 1.hours).toUnix
     var fetchedBytes: uint = 0
 
     let onBlocks = proc(
@@ -127,7 +126,7 @@ asyncchecksuite "Test Node - Host contracts":
         fetchedBytes += blk.data.len.uint
       return success()
 
-    (await onStore(request, 1.uint64, onBlocks, isRepairing = false)).tryGet()
+    (await onStore(request, expiry, 1.uint64, onBlocks, isRepairing = false)).tryGet()
     check fetchedBytes == 12 * DefaultBlockSize.uint
 
     let indexer = verifiable.protectedStrategy.init(
@@ -141,4 +140,4 @@ asyncchecksuite "Test Node - Host contracts":
         bytes = (await localStoreMetaDs.get(key)).tryGet
         blkMd = BlockMetadata.decode(bytes).tryGet
 
-      check blkMd.expiry == request.expiry.toSecondsSince1970
+      check blkMd.expiry == expiry