Add a basic ContentDB for portal networks (#848)

* Add a basic ContentDB for Portal networks

* Use ContentDB in StateNetwork

* Avoid probably some form of sandwich problem by re-exporting kvstore_sqlite3 from content_db

fluffy/conf.nim

@@ -8,6 +8,7 @@
 {.push raises: [Defect].}
 
 import
+ std/os,
  uri, confutils, confutils/std/net, chronicles,
  eth/keys, eth/net/nat, eth/p2p/discoveryv5/[enr, node],
  json_rpc/rpcproxy
@@ -60,6 +61,11 @@ type
  defaultValue: PrivateKey.random(keys.newRng()[])
  name: "nodekey" .}: PrivateKey
 
+ dataDir* {.
+ desc: "The directory where fluffy will store the content data"
+ defaultValue: config.defaultDataDir()
+ name: "data-dir" }: OutDir
+
  # Note: This will add bootstrap nodes for each enabled Portal network.
  # No distinction is being made on bootstrap nodes for a specific network.
  portalBootnodes* {.
@@ -164,3 +170,13 @@ proc parseCmdArg*(T: type ClientConfig, p: TaintedString): T
 
 proc completeCmdArg*(T: type ClientConfig, val: TaintedString): seq[string] =
  return @[]
+
+proc defaultDataDir*(config: PortalConf): string =
+ let dataDir = when defined(windows):
+ "AppData" / "Roaming" / "Fluffy"
+ elif defined(macosx):
+ "Library" / "Application Support" / "Fluffy"
+ else:
+ ".cache" / "fluffy"
+
+ getHomeDir() / dataDir

fluffy/content_db.nim

@@ -0,0 +1,87 @@
+# Nimbus
+# Copyright (c) 2021 Status Research & Development GmbH
+# Licensed and distributed under either of
+# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
+# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+{.push raises: [Defect].}
+
+import
+ std/options,
+ eth/db/kvstore,
+ eth/db/kvstore_sqlite3,
+ stint,
+ ./network/state/state_content
+
+export kvstore_sqlite3
+
+# This version of content db is the most basic, simple solution where data is
+# stored no matter what content type or content network in the same kvstore with
+# the content id as key. The content id is derived from the content key, and the
+# deriviation is different depending on the content type. As we use content id,
+# this part is currently out of the scope / API of the ContentDB.
+# In the future it is likely that that either:
+# 1. More kvstores are added per network, and thus depending on the network a
+# different kvstore needs to be selected.
+# 2. Or more kvstores are added per network and per content type, and thus
+# content key fields are required to access the data.
+# 3. Or databases are created per network (and kvstores pre content type) and
+# thus depending on the network the right db needs to be selected.
+
+type
+ ContentDB* = ref object
+ kv: KvStoreRef
+
+template expectDb(x: auto): untyped =
+ # There's no meaningful error handling implemented for a corrupt database or
+ # full disk - this requires manual intervention, so we'll panic for now
+ x.expect("working database (disk broken/full?)")
+
+proc new*(T: type ContentDB, path: string, inMemory = false): ContentDB =
+ let db =
+ if inMemory:
+ SqStoreRef.init("", "fluffy-test", inMemory = true).expect(
+ "working database (out of memory?)")
+ else:
+ SqStoreRef.init(path, "fluffy").expectDb()
+
+ ContentDB(kv: kvStore db.openKvStore().expectDb())
+
+proc get*(db: ContentDB, key: openArray[byte]): Option[seq[byte]] =
+ var res: Option[seq[byte]]
+ proc onData(data: openArray[byte]) = res = some(@data)
+
+ discard db.kv.get(key, onData).expectDb()
+
+ return res
+
+proc put*(db: ContentDB, key, value: openArray[byte]) =
+ db.kv.put(key, value).expectDb()
+
+proc contains*(db: ContentDB, key: openArray[byte]): bool =
+ db.kv.contains(key).expectDb()
+
+proc del*(db: ContentDB, key: openArray[byte]) =
+ db.kv.del(key).expectDb()
+
+# TODO: Could also decide to use the ContentKey SSZ bytestring, as this is what
+# gets send over the network in requests, but that would be a bigger key. Or the
+# same hashing could be done on it here.
+# However ContentId itself is already derived through different digests
+# depending on the content type, and this ContentId typically needs to be
+# checked with the Radius/distance of the node anyhow. So lets see how we end up
+# using this mostly in the code.
+
+proc get*(db: ContentDB, key: ContentId): Option[seq[byte]] =
+ # TODO: Here it is unfortunate that ContentId is a uint256 instead of Digest256.
+ db.get(key.toByteArrayBE())
+
+proc put*(db: ContentDB, key: ContentId, value: openArray[byte]) =
+ db.put(key.toByteArrayBE(), value)
+
+proc contains*(db: ContentDB, key: ContentId): bool =
+ db.contains(key.toByteArrayBE())
+
+proc del*(db: ContentDB, key: ContentId) =
+ db.del(key.toByteArrayBE())

fluffy/fluffy.nim

@@ -8,13 +8,16 @@
 {.push raises: [Defect].}
 
 import
+ std/os,
  confutils, confutils/std/net, chronicles, chronicles/topics_registry,
  chronos, metrics, metrics/chronos_httpserver, json_rpc/clients/httpclient,
- json_rpc/rpcproxy,
+ json_rpc/rpcproxy, stew/byteutils,
  eth/keys, eth/net/nat,
  eth/p2p/discoveryv5/protocol as discv5_protocol,
+ eth/p2p/discoveryv5/node,
  ./conf, ./rpc/[eth_api, bridge_client, discovery_api],
- ./network/state/[state_network, state_content]
+ ./network/state/[state_network, state_content],
+ ./content_db
 
 proc initializeBridgeClient(maybeUri: Option[string]): Option[BridgeClient] =
  try:
@@ -53,7 +56,14 @@ proc run(config: PortalConf) {.raises: [CatchableError, Defect].} =
 
  d.open()
 
- let stateNetwork = StateNetwork.new(d, newEmptyInMemoryStorage(),
+ # Store the database at contentdb prefixed with the first 8 chars of node id.
+ # This is done because the content in the db is dependant on the `NodeId` and
+ # the selected `Radius`.
+ let db =
+ ContentDB.new(config.dataDir / "db" / "contentdb_" &
+ d.localNode.id.toByteArrayBE().toOpenArray(0, 8).toHex())
+
+ let stateNetwork = StateNetwork.new(d, db,
  bootstrapRecords = config.portalBootnodes)
 
  if config.metricsEnabled:

fluffy/network/state/state_network.nim

@@ -2,6 +2,7 @@ import
  std/[options, sugar],
  stew/[results, byteutils],
  eth/p2p/discoveryv5/[protocol, node, enr],
+ ../../content_db,
  ../wire/portal_protocol,
  ./state_content
 
@@ -12,15 +13,16 @@ const
 # objects i.e nodes, tries, hashes
 type StateNetwork* = ref object
  portalProtocol*: PortalProtocol
- storage: ContentStorage
+ contentDB*: ContentDB
 
-proc getHandler(storage: ContentStorage): ContentHandler =
+proc getHandler(contentDB: ContentDB): ContentHandler =
  return (proc (contentKey: state_content.ByteList): ContentResult =
- let maybeContent = storage.get(contentKey)
+ let contentId = toContentId(contentKey)
+ let maybeContent = contentDB.get(contentId)
  if (maybeContent.isSome()):
  ContentResult(kind: ContentFound, content: maybeContent.unsafeGet())
  else:
- ContentResult(kind: ContentMissing, contentId: toContentId(contentKey)))
+ ContentResult(kind: ContentMissing, contentId: contentId))
 
 # Further improvements which may be necessary:
 # 1. Return proper domain types instead of bytes
@@ -37,13 +39,13 @@ proc getContent*(p: StateNetwork, key: ContentKey):
  return content.map(x => x.asSeq())
 
 proc new*(T: type StateNetwork, baseProtocol: protocol.Protocol,
- storage: ContentStorage , dataRadius = UInt256.high(),
+ contentDB: ContentDB , dataRadius = UInt256.high(),
  bootstrapRecords: openarray[Record] = []): T =
  let portalProtocol = PortalProtocol.new(
- baseProtocol, StateProtocolId, getHandler(storage), dataRadius,
+ baseProtocol, StateProtocolId, getHandler(contentDB), dataRadius,
  bootstrapRecords)
 
- return StateNetwork(portalProtocol: portalProtocol, storage: storage)
+ return StateNetwork(portalProtocol: portalProtocol, contentDB: contentDB)
 
 proc start*(p: StateNetwork) =
  p.portalProtocol.start()

fluffy/tests/all_fluffy_tests.nim

@@ -12,5 +12,6 @@ import
  ./test_portal_wire_protocol,
  ./test_custom_distance,
  ./test_state_network,
+ ./test_content_db,
  ./test_discovery_rpc,
  ./test_bridge_parser

fluffy/tests/test_content_db.nim

@@ -0,0 +1,45 @@
+# Nimbus
+# Copyright (c) 2021 Status Research & Development GmbH
+# Licensed and distributed under either of
+# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
+# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+{.used.}
+
+import
+ unittest2, stint,
+ ../network/state/state_content,
+ ../content_db
+
+suite "Content Database":
+ # Note: We are currently not really testing something new here just basic
+ # underlying kvstore.
+ test "ContentDB basic API":
+ let
+ db = ContentDB.new("", inMemory = true)
+ key = ContentId(UInt256.high()) # Some key
+
+ block:
+ let val = db.get(key)
+
+ check:
+ val.isNone()
+ db.contains(key) == false
+
+ block:
+ db.put(key, [byte 0, 1, 2, 3])
+ let val = db.get(key)
+
+ check:
+ val.isSome()
+ val.get() == [byte 0, 1, 2, 3]
+ db.contains(key) == true
+
+ block:
+ db.del(key)
+ let val = db.get(key)
+
+ check:
+ val.isNone()
+ db.contains(key) == false

fluffy/tests/test_state_network.nim

@@ -13,6 +13,7 @@ import
  ../../nimbus/[genesis, chain_config, config, db/db_chain],
  ../network/wire/portal_protocol,
  ../network/state/[state_content, state_network],
+ ../content_db,
  ./test_helpers
 
 proc genesisToTrie(filePath: string): HexaryTrie =
@@ -45,15 +46,27 @@ procSuite "State Content Network":
  node2 = initDiscoveryNode(
  rng, PrivateKey.random(rng[]), localAddress(20303))
 
- proto1 = StateNetwork.new(node1, ContentStorage(trie: trie))
- proto2 = StateNetwork.new(node2, ContentStorage(trie: trie))
+ proto1 = StateNetwork.new(node1, ContentDB.new("", inMemory = true))
+ proto2 = StateNetwork.new(node2, ContentDB.new("", inMemory = true))
 
  check proto2.portalProtocol.addNode(node1.localNode) == Added
 
  var keys: seq[seq[byte]]
  for k, v in trie.replicate:
  keys.add(k)
 
+ var nodeHash: NodeHash
+ copyMem(nodeHash.data.addr, unsafeAddr k[0], sizeof(nodeHash.data))
+
+ let
+ contentKey = ContentKey(
+ networkId: 0'u16,
+ contentType: state_content.ContentType.Account,
+ nodeHash: nodeHash)
+ contentId = toContentId(contentKey)
+
+ proto1.contentDB.put(contentId, v)
+
  for key in keys:
  var nodeHash: NodeHash
  copyMem(nodeHash.data.addr, unsafeAddr key[0], sizeof(nodeHash.data))
@@ -90,9 +103,9 @@ procSuite "State Content Network":
  rng, PrivateKey.random(rng[]), localAddress(20304))
 
 
- proto1 = StateNetwork.new(node1, ContentStorage(trie: trie))
- proto2 = StateNetwork.new(node2, ContentStorage(trie: trie))
- proto3 = StateNetwork.new(node3, ContentStorage(trie: trie))
+ proto1 = StateNetwork.new(node1, ContentDB.new("", inMemory = true))
+ proto2 = StateNetwork.new(node2, ContentDB.new("", inMemory = true))
+ proto3 = StateNetwork.new(node3, ContentDB.new("", inMemory = true))
 
 
  # Node1 knows about Node2, and Node2 knows about Node3 which hold all content
@@ -105,6 +118,21 @@ procSuite "State Content Network":
  for k, v in trie.replicate:
  keys.add(k)
 
+ var nodeHash: NodeHash
+ copyMem(nodeHash.data.addr, unsafeAddr k[0], sizeof(nodeHash.data))
+
+ let
+ contentKey = ContentKey(
+ networkId: 0'u16,
+ contentType: state_content.ContentType.Account,
+ nodeHash: nodeHash)
+ contentId = toContentId(contentKey)
+
+ proto2.contentDB.put(contentId, v)
+ # Not needed right now as 1 node is enough considering node 1 is connected
+ # to both.
+ proto3.contentDB.put(contentId, v)
+
  # Get first key
  var nodeHash: NodeHash
  let firstKey = keys[0]