Handle quota and cleanup correctly

.dockerignore

@@ -4,3 +4,5 @@ docs
 metrics
 nimcache
 tests
+vendor/nimble/nat_traversal/vendor/miniupnp/miniupnpc/build
+vendor/nimble/nat_traversal/vendor/libnatpmp-upstream/libnatpmp.a

.github/workflows/docker-dist-tests.yml

@@ -28,6 +28,11 @@ on:
         required: false
         type: boolean
         default: false
+      release_tests_namespace:
+        description: Kubernetes namespace for release tests
+        required: false
+        type: string
+        default: 'devnet'
 
 jobs:
   get-contracts-hash:
@@ -57,6 +62,7 @@ jobs:
       tag_stable: ${{ startsWith(github.ref, 'refs/tags/') }}
       contract_image: "durabilitylabs/archivist-contracts:sha-${{ needs.get-contracts-hash.outputs.hash }}-dist-tests"
       run_release_tests: ${{ inputs.run_release_tests }}
+      release_tests_namespace: ${{ inputs.release_tests_namespace || 'devnet' }}
     secrets: inherit
 
   deploy-testnet:

.github/workflows/docker-reusable.yml

@@ -49,6 +49,11 @@ on:
         required: false
         type: string
         default: false
+      release_tests_namespace:
+        description: Kubernetes namespace for release tests
+        required: false
+        type: string
+        default: 'devnet'
       contract_image:
         description: Specifies compatible smart contract image
         required: false
@@ -291,5 +296,6 @@ jobs:
       branch: ${{ needs.compute-tests-inputs.outputs.branch }}
       archivistdockerimage: ${{ needs.compute-tests-inputs.outputs.archivistdockerimage }}
       workflow_source: ${{ needs.compute-tests-inputs.outputs.workflow_source }}
+      namespace: ${{ inputs.release_tests_namespace || 'devnet' }}
     secrets:
       KUBE_CONFIG: ${{ secrets.DEVNET_KUBE_CONFIG }}

archivist/archivist.nim

@@ -216,11 +216,8 @@ proc new*(
       overlayTtl = config.overlayTtl.seconds,
     )
 
-    maintenance = BlockMaintainer.new(
-      repoStore,
-      interval = config.overlayMaintenanceInterval,
-      numberOfBlocksPerInterval = config.overlayMaintenanceNumberOfBlocks,
-    )
+    maintenance =
+      BlockMaintainer.new(repoStore, interval = config.overlayMaintenanceInterval)
 
     peerStore = PeerCtxStore.new()
     pendingBlocks = PendingBlocksManager.new()

archivist/archivisttypes.nim

@@ -9,6 +9,7 @@
 
 {.push raises: [].}
 
+import std/math
 import std/tables
 import std/sugar
 
@@ -25,13 +26,13 @@ export tables
 
 const
   # Size of blocks for storage / network exchange,
-  DefaultBlockSize* = NBytes 1024 * 64
-  DefaultCellSize* = NBytes 2048
+  DefaultBlockSize* = 512.KiBs
+  DefaultCellSize* = 2.KiBs
 
   # Proving defaults
   DefaultMaxSlotDepth* = 32
   DefaultMaxDatasetDepth* = 8
-  DefaultBlockDepth* = 5
+  DefaultBlockDepth* = int(log2(float(DefaultBlockSize div DefaultCellSize)))
   DefaultCellElms* = 67
   DefaultSamplesNum* = 5
 

archivist/blockexchange/engine/engine.nim

@@ -451,10 +451,18 @@ proc blocksDeliveryHandler*(
       validatedBlocksDelivery.keepItIf(it.address.cid != bd.address.cid)
       continue
 
-    # TODO: The putBlock here should be replace by something like -
-    # storeManifestBlock(...)
-    if err =? (await self.localStore.putBlock(bd.blk)).errorOption:
-      error "Unable to store block", err = err.msg
+    if not isManifest:
+      error "Skipping, blocks should either be part of a tree or a manifest",
+        cid = bd.blk.cid
+      continue
+
+    without manifest =? Manifest.decode(bd.blk), err:
+      error "Unable to decode manifest block", err = err.msg
+      validatedBlocksDelivery.keepItIf(it.address.cid != bd.address.cid)
+      continue
+
+    if err =? (await self.localStore.storeManifest(manifest)).errorOption:
+      error "Unable to store manifest", err = err.msg
       validatedBlocksDelivery.keepItIf(it.address.cid != bd.address.cid)
 
   archivist_block_exchange_blocks_received.inc(validatedBlocksDelivery.len.int64)

archivist/clock.nim

@@ -30,6 +30,19 @@ proc withTimeout*(
     await future.cancelAndWait()
     raise newException(Timeout, "Timed out")
 
+proc withTimeout*[T](
+    future: Future[T], clock: Clock, expiry: SecondsSince1970
+): Future[T] {.async.} =
+  let timeout = clock.waitUntil(expiry)
+  try:
+    await future or timeout
+  finally:
+    await timeout.cancelAndWait()
+  if not future.completed:
+    await future.cancelAndWait()
+    raise newException(Timeout, "Timed out")
+  return future.read()
+
 proc toBytes*(i: SecondsSince1970): seq[byte] =
   let asUint = cast[uint64](i)
   @(asUint.toBytes)

archivist/erasure/erasure.nim

@@ -634,6 +634,12 @@ proc decode*(
   ##             be recovered
   ##
 
+  logScope:
+    treeCid = encoded.treeCid
+    originalTreeCid = encoded.originalTreeCid
+
+  trace "Preparing to decode dataset"
+
   let
     emptyCid = ?emptyCid(encoded.version, encoded.hcodec, encoded.codec)
     params = EncodingParams.initFromEncoded(encoded, emptyCid)
@@ -663,6 +669,7 @@ proc decode*(
       if err =? (await self.repoStore.putSomeProofs(tree, idxIter)).errorOption:
         return failure(err)
 
+      trace "Successfully decoded original dataset"
       success(),
   )
 
@@ -689,6 +696,12 @@ proc repair*(
   ## to avoid leaving garbage in the store
   ##
 
+  logScope:
+    treeCid = encoded.treeCid
+    originalTreeCid = encoded.originalTreeCid
+
+  trace "Preparing to repair dataset"
+
   let
     emptyCid = ?emptyCid(encoded.version, encoded.hcodec, encoded.codec)
     params = EncodingParams.initFromEncoded(encoded, emptyCid)
@@ -708,8 +721,9 @@ proc repair*(
           "Original tree root differs from the tree root computed out of recovered data"
         )
 
-      await self.repoStore.putAllProofs(tree)
-    ,
+      ?await self.repoStore.putAllProofs(tree)
+      trace "Successfully repaired original dataset"
+      success(),
   )
 
   # TODO: We don't get valid parity data from leopard,
@@ -719,6 +733,7 @@ proc repair*(
   let repaired =
     ?(await self.encode(encoded, encoded.ecK, encoded.ecM, encoded.protectedStrategy))
 
+  trace "Successfully re-encoded original dataset"
   if repaired.treeCid != encoded.treeCid:
     return failure(
       "Original tree root differs from the repaired tree root encoded out of recovered data"

archivist/marketplace/sales/states/filled.nim

@@ -55,10 +55,18 @@ method run*(
       if onFilled =? agent.onFilled:
         onFilled(request, data.slotIndex)
 
+      # Add buffer past contract end so overlay data survives through the
+      # last proof window. Buffer covers period (last proof window) +
+      # proofTimeout (challenge window). Without this, maintenance can drop
+      # the overlay while the proving loop is still running its last proof.
       let expiry = await marketplace.getRequestEnd(data.requestId)
+      let periodicity = marketplace.periodicity()
+      let buffer = periodicity.seconds + marketplace.proofTimeout()
+      let expiryWithBuffer = expiry + buffer
       let cid = request.content.cid
       let slotIndex = data.slotIndex
-      if err =? (await storage.updateSlotExpiry(cid, slotIndex, expiry)).errorOption:
+      if err =?
+          (await storage.updateSlotExpiry(cid, slotIndex, expiryWithBuffer)).errorOption:
         return some State(SaleErrored(error: err))
 
       when defined(archivist_system_testing_options):

archivist/marketplace/sales/states/initialproving.nim

@@ -70,10 +70,14 @@ method run*(
     let cid = request.content.cid
     let slotIndex = data.slotIndex
     let challenge = await context.marketplace.getChallenge(slot.id)
-    without proof =? await storage.proveSlot(cid, slotIndex, challenge), err:
+    # Deadline is periodEnd - proof must be generated before the period closes.
+    let periodEnd = periodicity.periodEnd(provingPeriod).toSecondsSince1970
+
+    without proof =?
+      (await storage.proveSlot(cid, slotIndex, challenge).withTimeout(clock, periodEnd)),
+      err:
       error "Failed to generate initial proof", error = err.msg
       return some State(SaleErrored(error: err))
-
     let periodAtFinish = periodicity.periodOf(StorageTimestamp.init(clock.now()))
     if periodAtFinish != provingPeriod:
       warn "Failed to generate initial proof in time",

archivist/marketplace/sales/states/proving.nim

@@ -1,5 +1,6 @@
 import std/options
 import pkg/questionable/results
+import pkg/chronos
 import ../../../clock
 import ../../../logutils
 import ../../../utils/exceptions
@@ -85,7 +86,18 @@ proc proveLoop(
           (await marketplace.willProofBeRequired(slotId)):
         let challenge = await marketplace.getChallenge(slotId)
         info "Generating required proof", challenge = challenge
-        await state.prove(slot, challenge, context, provingPeriod)
+        # Deadline is the end of the current proving period. Proof must be
+        # generated and submitted before periodEnd.
+        let periodicity = marketplace.periodicity()
+        let periodEnd = periodicity.periodEnd(provingPeriod).toSecondsSince1970
+        try:
+          await state.prove(slot, challenge, context, provingPeriod).withTimeout(
+            clock, periodEnd
+          )
+        except Timeout:
+          warn "Proof generation timed out", provingPeriod = provingPeriod
+          await waitUntilPeriod(provingPeriod + 1'u8)
+          continue
         let periodAtFinish = getCurrentPeriod()
         if periodAtFinish != provingPeriod:
           warn "Failed to generate proof in time", periodAtFinish = periodAtFinish
@@ -123,7 +135,6 @@ method run*(
 ): Future[?State] {.async: (raises: []).} =
   let data = SalesAgent(machine).data
   let context = SalesAgent(machine).context
-  let clock = context.clock
 
   without request =? data.request:
     raiseAssert "no sale request"

archivist/marketplacestorage.nim

@@ -35,8 +35,7 @@ method proveSlot*(
 method updateSlotExpiry*(
     storage: MarketplaceStorage, cid: Cid, slotIndex: uint64, expiry: StorageTimestamp
 ): Future[?!void] {.async: (raises: [CancelledError]).} =
-  # TODO: update only the slot expiry, not the expiry of the entiry dataset
-  await storage.node.updateExpiry(cid, expiry.toSecondsSince1970)
+  await storage.node.updateSlotExpiry(cid, slotIndex, expiry.toSecondsSince1970)
 
 method deleteSlot*(
     storage: MarketplaceStorage, cid: Cid, slotIndex: uint64