Expiry per dataset

Author: tbekas

README.md

@@ -88,12 +88,9 @@ The following options are available:
      --api-bindaddr         The REST API bind address [=127.0.0.1].
  -p, --api-port             The REST Api port [=8080].
      --repo-kind            Backend for main repo store (fs, sqlite) [=fs].
- -q, --storage-quota        The size of the total storage quota dedicated to the node [=8589934592].
- -t, --block-ttl            Default block timeout in seconds - 0 disables the ttl [=$DefaultBlockTtl].
-     --block-mi             Time interval in seconds - determines frequency of block maintenance cycle: how
-                            often blocks are checked for expiration and cleanup
-                            [=$DefaultBlockMaintenanceInterval].
-     --block-mn             Number of blocks to check every maintenance cycle [=1000].
+ -q, --storage-quota        The size in bytes of the total storage quota dedicated to the node [=8589934592].
+ -t, --default-ttl          Default dataset expiry [=1d].
+     --maintenance-interval Determines how frequently datasets are checked for expiration and cleanup [=5m].
  -c, --cache-size           The size of the block cache, 0 disables the cache - might help on slow hardrives
                             [=0].
 

codex/codex.nim

@@ -54,7 +54,7 @@ type
     restServer: RestServerRef
     codexNode: CodexNodeRef
     repoStore: RepoStore
-    maintenance: BlockMaintainer
+    maintenance: DatasetMaintainer
     taskpool: Taskpool
 
   CodexPrivateKey* = libp2p.PrivateKey # alias
@@ -246,6 +246,9 @@ proc new*(
     wallet = WalletRef.new(EthPrivateKey.random())
     network = BlockExcNetwork.new(switch)
 
+    metaDs = LevelDbDatastore.new(config.dataDir / CodexMetaNamespace)
+        .expect("Should create metadata store!")
+
     repoData = case config.repoKind
                 of repoFS: Datastore(FSDatastore.new($config.dataDir, depth = 5)
                   .expect("Should create repo file data store!"))
@@ -256,21 +259,21 @@ proc new*(
 
     repoStore = RepoStore.new(
       repoDs = repoData,
-      metaDs = LevelDbDatastore.new(config.dataDir / CodexMetaNamespace)
-        .expect("Should create metadata store!"),
-      quotaMaxBytes = config.storageQuota,
-      blockTtl = config.blockTtl)
+      metaDs = metaDs,
+      quotaMaxBytes = config.storageQuota)
 
-    maintenance = BlockMaintainer.new(
+    maintenance = DatasetMaintainer.new(
       repoStore,
-      interval = config.blockMaintenanceInterval,
-      numberOfBlocksPerInterval = config.blockMaintenanceNumberOfBlocks)
+      metaDs,
+      defaultExpiry = config.defaultExpiry,
+      interval = config.maintenanceInterval)
 
     peerStore = PeerCtxStore.new()
     pendingBlocks = PendingBlocksManager.new()
     blockDiscovery = DiscoveryEngine.new(repoStore, peerStore, network, discovery, pendingBlocks)
     engine = BlockExcEngine.new(repoStore, wallet, network, blockDiscovery, peerStore, pendingBlocks)
     store = NetworkStore.new(engine, repoStore)
+
     prover = if config.prover:
       if not fileAccessible($config.circomR1cs, {AccessFlags.Read}) and
         endsWith($config.circomR1cs, ".r1cs"):
@@ -306,6 +309,7 @@ proc new*(
     codexNode = CodexNodeRef.new(
       switch = switch,
       networkStore = store,
+      maintenance = maintenance,
       engine = engine,
       prover = prover,
       discovery = discovery,

codex/conf.nim

@@ -42,9 +42,8 @@ export units, net, codextypes, logutils
 
 export
   DefaultQuotaBytes,
-  DefaultBlockTtl,
-  DefaultBlockMaintenanceInterval,
-  DefaultNumberOfBlocksToMaintainPerInterval
+  DefaultDefaultExpiry,
+  DefaultMaintenanceInterval
 
 proc defaultDataDir*(): string =
   let dataDir = when defined(windows):
@@ -209,24 +208,18 @@ type
       name: "storage-quota"
       abbr: "q" }: NBytes
 
-    blockTtl* {.
-      desc: "Default block timeout in seconds - 0 disables the ttl"
-      defaultValue: DefaultBlockTtl
-      defaultValueDesc: $DefaultBlockTtl
-      name: "block-ttl"
+    defaultExpiry* {.
+      desc: "Default dataset expiry in seconds"
+      defaultValue: DefaultDefaultExpiry
+      defaultValueDesc: $DefaultDefaultExpiry
+      name: "default-ttl"
       abbr: "t" }: Duration
 
-    blockMaintenanceInterval* {.
-      desc: "Time interval in seconds - determines frequency of block maintenance cycle: how often blocks are checked for expiration and cleanup"
-      defaultValue: DefaultBlockMaintenanceInterval
-      defaultValueDesc: $DefaultBlockMaintenanceInterval
-      name: "block-mi" }: Duration
-
-    blockMaintenanceNumberOfBlocks* {.
-      desc: "Number of blocks to check every maintenance cycle"
-      defaultValue: DefaultNumberOfBlocksToMaintainPerInterval
-      defaultValueDesc: $DefaultNumberOfBlocksToMaintainPerInterval
-      name: "block-mn" }: int
+    maintenanceInterval* {.
+      desc: "Determines how frequently datasets are checked for expiration and cleanup"
+      defaultValue: DefaultMaintenanceInterval
+      defaultValueDesc: $DefaultMaintenanceInterval
+      name: "maintenance-interval" }: Duration
 
     cacheSize* {.
       desc: "The size of the block cache, 0 disables the cache - might help on slow hardrives"

codex/namespaces.nim

@@ -18,6 +18,8 @@ const
     CodexMetaNamespace & "/ttl"
   CodexBlockProofNamespace* =                                  # Cid and Proof
     CodexMetaNamespace & "/proof"
+  CodexDatasetMetadataNamespace* =                                  # Dataset
+    CodexMetaNamespace & "/dataset"
   CodexDhtNamespace* = "dht"                                  # Dht namespace
   CodexDhtProvidersNamespace* =                               # Dht providers namespace
     CodexDhtNamespace & "/providers"

codex/node.nim

@@ -65,6 +65,7 @@ type
     switch: Switch
     networkId: PeerId
     networkStore: NetworkStore
+    maintenance: DatasetMaintainer
     engine: BlockExcEngine
     prover: ?Prover
     discovery: Discovery
@@ -155,17 +156,8 @@ proc updateExpiry*(
     trace "Unable to fetch manifest for cid", manifestCid
     return failure(error)
 
-  try:
-    let
-      ensuringFutures = Iter[int].new(0..<manifest.blocksCount)
-        .mapIt(self.networkStore.localStore.ensureExpiry( manifest.treeCid, it, expiry ))
-    await allFuturesThrowing(ensuringFutures)
-  except CancelledError as exc:
-    raise exc
-  except CatchableError as exc:
-    return failure(exc.msg)
+  await self.maintenance.ensureExpiry(manifest.treeCid, expiry)
 
-  return success()
 
 proc fetchBatched*(
   self: CodexNodeRef,
@@ -274,6 +266,13 @@ proc streamEntireDataset(
 
   # Retrieve all blocks of the dataset sequentially from the local store or network
   trace "Creating store stream for manifest", manifestCid
+
+  if err =? (await self.maintenance.trackExpiry(
+      manifest.treeCid,
+      manifest.blocksCount,
+      manifestsCids = @[manifestCid])).errorOption:
+    return failure(err)
+
   LPStream(StoreStream.new(self.networkStore, manifest, pad = false)).success
 
 proc retrieve*(
@@ -361,6 +360,12 @@ proc store*(
     error "Unable to store manifest"
     return failure(err)
 
+  if err =? (await self.maintenance.trackExpiry(
+      treeCid,
+      manifest.blocksCount,
+      manifestsCids = @[manifestBlk.cid])).errorOption:
+    return failure(err)
+
   info "Stored data", manifestCid = manifestBlk.cid,
                       treeCid = treeCid,
                       blocks = manifest.blocksCount,
@@ -566,18 +571,17 @@ proc onStore(
     trace "Slot index not in manifest", slotIdx
     return failure(newException(CodexError, "Slot index not in manifest"))
 
-  proc updateExpiry(blocks: seq[bt.Block]): Future[?!void] {.async.} =
-    trace "Updating expiry for blocks", blocks = blocks.len
-
-    let ensureExpiryFutures = blocks.mapIt(self.networkStore.ensureExpiry(it.cid, expiry))
-    if updateExpiryErr =? (await allFutureResult(ensureExpiryFutures)).errorOption:
-      return failure(updateExpiryErr)
-
-    if not blocksCb.isNil and err =? (await blocksCb(blocks)).errorOption:
-      trace "Unable to process blocks", err = err.msg
-      return failure(err)
+  proc onBatch(blocks: seq[bt.Block]): Future[?!void] {.async.} =
+    if not blocksCb.isNil:
+      await blocksCb(blocks)
+    else:
+      success()
 
-    return success()
+  if err =? (await self.maintenance.trackExpiry(
+      manifest.treeCid,
+      manifest.blocksCount,
+      manifestsCids = @[cid])).errorOption:
+    return failure(err)
 
   without indexer =? manifest.verifiableStrategy.init(
     0, manifest.blocksCount - 1, manifest.numSlots).catch, err:
@@ -591,7 +595,7 @@ proc onStore(
   if err =? (await self.fetchBatched(
       manifest.treeCid,
       blksIter,
-      onBatch = updateExpiry)).errorOption:
+      onBatch = onBatch)).errorOption:
     trace "Unable to fetch blocks", err = err.msg
     return failure(err)
 
@@ -607,6 +611,11 @@ proc onStore(
 
   trace "Slot successfully retrieved and reconstructed"
 
+  if err =? (await self.maintenance.ensureExpiry(
+      manifest.treeCid,
+      expiry)).errorOption:
+    return failure(err)
+
   return success()
 
 proc onProve(
@@ -771,6 +780,7 @@ proc new*(
   T: type CodexNodeRef,
   switch: Switch,
   networkStore: NetworkStore,
+  maintenance: DatasetMaintainer,
   engine: BlockExcEngine,
   discovery: Discovery,
   prover = Prover.none,
@@ -782,6 +792,7 @@ proc new*(
   CodexNodeRef(
     switch: switch,
     networkStore: networkStore,
+    maintenance: maintenance,
     engine: engine,
     prover: prover,
     discovery: discovery,

codex/stores/blockstore.nim

@@ -61,9 +61,9 @@ method getBlockAndProof*(self: BlockStore, treeCid: Cid, index: Natural): Future
   raiseAssert("getBlockAndProof not implemented!")
 
 method putBlock*(
-  self: BlockStore,
-  blk: Block,
-  ttl = Duration.none): Future[?!void] {.base.} =
+    self: BlockStore,
+    blk: Block
+): Future[?!void] {.base.} =
   ## Put a block to the blockstore
   ##
 
@@ -89,27 +89,6 @@ method getCidAndProof*(
 
   raiseAssert("getCidAndProof not implemented!")
 
-method ensureExpiry*(
-  self: BlockStore,
-  cid: Cid,
-  expiry: SecondsSince1970): Future[?!void] {.base.} =
-  ## Ensure that block's assosicated expiry is at least given timestamp
-  ## If the current expiry is lower then it is updated to the given one, otherwise it is left intact
-  ##
-
-  raiseAssert("Not implemented!")
-
-method ensureExpiry*(
-  self: BlockStore,
-  treeCid: Cid,
-  index: Natural,
-  expiry: SecondsSince1970): Future[?!void] {.base.} =
-  ## Ensure that block's associated expiry is at least given timestamp
-  ## If the current expiry is lower then it is updated to the given one, otherwise it is left intact
-  ##
-
-  raiseAssert("Not implemented!")
-
 method delBlock*(self: BlockStore, cid: Cid): Future[?!void] {.base.} =
   ## Delete a block from the blockstore
   ##

codex/stores/cachestore.nim

@@ -186,8 +186,7 @@ func putBlockSync(self: CacheStore, blk: Block): bool =
 
 method putBlock*(
   self: CacheStore,
-  blk: Block,
-  ttl = Duration.none): Future[?!void] {.async.} =
+  blk: Block): Future[?!void] {.async.} =
   ## Put a block to the blockstore
   ##
 
@@ -209,27 +208,6 @@ method putCidAndProof*(
   self.cidAndProofCache[(treeCid, index)] = (blockCid, proof)
   success()
 
-method ensureExpiry*(
-    self: CacheStore,
-    cid: Cid,
-    expiry: SecondsSince1970
-): Future[?!void] {.async.} =
-  ## Updates block's assosicated TTL in store - not applicable for CacheStore
-  ##
-
-  discard # CacheStore does not have notion of TTL
-
-method ensureExpiry*(
-    self: CacheStore,
-    treeCid: Cid,
-    index: Natural,
-    expiry: SecondsSince1970
-): Future[?!void] {.async.} =
-  ## Updates block's associated TTL in store - not applicable for CacheStore
-  ##
-
-  discard # CacheStore does not have notion of TTL
-
 method delBlock*(self: CacheStore, cid: Cid): Future[?!void] {.async.} =
   ## Delete a block from the blockstore
   ##

codex/stores/keyutils.nim

@@ -25,6 +25,7 @@ const
   CodexManifestKey* = Key.init(CodexManifestNamespace).tryGet
   BlocksTtlKey* = Key.init(CodexBlocksTtlNamespace).tryGet
   BlockProofKey* = Key.init(CodexBlockProofNamespace).tryGet
+  DatasetMetadataKey* = Key.init(CodexDatasetMetadataNamespace).tryGet
   QuotaKey* = Key.init(CodexQuotaNamespace).tryGet
   QuotaUsedKey* = (QuotaKey / "used").tryGet
   QuotaReservedKey* = (QuotaKey / "reserved").tryGet
@@ -47,3 +48,10 @@ proc createBlockExpirationMetadataQueryKey*(): ?!Key =
 
 proc createBlockCidAndProofMetadataKey*(treeCid: Cid, index: Natural): ?!Key =
   (BlockProofKey / $treeCid).flatMap((k: Key) => k / $index)
+
+proc createDatasetMetadataKey*(treeCid: Cid): ?!Key =
+  DatasetMetadataKey / $treeCid
+
+proc createDatasetMetadataQueryKey*(): ?!Key =
+  let queryString = ? (DatasetMetadataKey / "*")
+  Key.init(queryString)