fix(sales): set correct block expiry

Previously we used request.expiry as the block expiry,
but the former is a duration in seconds, whereas the
latter is a timestamp

Author: markspanbroek

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/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/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

codex/sales/states/downloading.nim

@@ -38,6 +38,7 @@ method run*(
   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 @@ method run*(
     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 @@ from pkg/ethers import BlockTag
 import codex/clock
 
 import ../examples
+import ./mockclock
 
 export market
 export tables
@@ -51,7 +52,7 @@ type
     errorOnFillSlot*: ?(ref MarketError)
     errorOnFreeSlot*: ?(ref MarketError)
     errorOnGetHost*: ?(ref MarketError)
-    clock: ?Clock
+    clock: Clock
 
   Fulfillment* = object
     requestId*: RequestId
@@ -63,7 +64,7 @@ type
     host*: Address
     slotIndex*: uint64
     proof*: Groth16Proof
-    timestamp: ?SecondsSince1970
+    timestamp: SecondsSince1970
     collateral*: UInt256
 
   Subscriptions = object
@@ -119,7 +120,7 @@ proc hash*(address: Address): Hash =
 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 getPointer*(market: MockMarket, slotId: SlotId): Future[uint8] {.async.}
 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 @@ proc fillSlot*(
     slotIndex: slotIndex,
     proof: proof,
     host: host,
-    timestamp: market.clock .? now,
+    timestamp: market.clock.now,
     collateral: collateral,
   )
   market.filled.add(slot)
@@ -541,15 +547,23 @@ method queryPastStorageRequestedEvents*(
 ): 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 @@ method queryPastSlotFilledEvents*(
 ): 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