New proto

beacon_chain/beacon_chain_db.nim

@@ -12,14 +12,16 @@ type
     kHashToState
     kHashToBlock
     kHeadBlock
+    kSlotToBlockRoots
 
 func subkey(kind: DbKeyKind): array[1, byte] =
   result[0] = byte ord(kind)
 
-func subkey[N: static int](kind: DbKeyKind, key: array[N, byte]):
-    array[N + 1, byte] =
-  result[0] = byte ord(kind)
-  result[1 .. ^1] = key
+func subkey[T](kind: DbKeyKind, key: T): auto =
+  var res: array[sizeof(T) + 1, byte]
+  res[0] = byte ord(kind)
+  copyMem(addr res[1], unsafeAddr key, sizeof(key))
+  return res
 
 func subkey(kind: type BeaconState, key: Eth2Digest): auto =
   subkey(kHashToState, key.data)
@@ -31,8 +33,23 @@ proc init*(T: type BeaconChainDB, backend: TrieDatabaseRef): BeaconChainDB =
   new result
   result.backend = backend
 
+proc toSeq(v: openarray[byte], ofType: type): seq[ofType] =
+  if v.len != 0:
+    assert(v.len mod sizeof(ofType) == 0)
+    let sz = v.len div sizeof(ofType)
+    result = newSeq[ofType](sz)
+    copyMem(addr result[0], unsafeAddr v[0], v.len)
+
 proc putBlock*(db: BeaconChainDB, key: Eth2Digest, value: BeaconBlock) =
-  db.backend.put(subkey(type value, key), ssz.serialize(value))
+  let slotKey = subkey(kSlotToBlockRoots, value.slot)
+  var blockRootsBytes = db.backend.get(slotKey)
+  var blockRoots = blockRootsBytes.toSeq(Eth2Digest)
+  if key notin blockRoots:
+    db.backend.put(subkey(type value, key), ssz.serialize(value))
+    blockRootsBytes.setLen(blockRootsBytes.len + sizeof(key))
+    copyMem(addr blockRootsBytes[^sizeof(key)], unsafeAddr key, sizeof(key))
+    db.backend.put(slotKey, blockRootsBytes)
+
 
 proc putHead*(db: BeaconChainDB, key: Eth2Digest) =
   db.backend.put(subkey(kHeadBlock), key.data) # TODO head block?
@@ -81,6 +98,9 @@ proc getHead*(db: BeaconChainDB): Option[BeaconBlock] =
   else:
     none(BeaconBlock)
 
+proc getBlockRootsForSlot*(db: BeaconChainDB, slot: uint64): seq[Eth2Digest] =
+  db.backend.get(subkey(kSlotToBlockRoots, slot)).toSeq(Eth2Digest)
+
 proc containsBlock*(
     db: BeaconChainDB, key: Eth2Digest): bool =
   db.backend.contains(subkey(BeaconBlock, key))

beacon_chain/beacon_node.nim

@@ -30,6 +30,8 @@ const
 
   stateStoragePeriod = SLOTS_PER_EPOCH.uint64 * 10 # Save states once per this number of slots. TODO: Find a good number.
 
+proc onBeaconBlock*(node: BeaconNode, blck: BeaconBlock)
+
 import sync_protocol
 
 func shortHash(x: auto): string =
@@ -525,7 +527,7 @@ proc updateHeadBlock(node: BeaconNode, blck: BeaconBlock) =
     headBlockRoot = shortHash(node.headBlockRoot),
     stateSlot = humaneSlotNum(node.beaconState.slot)
 
-proc onBeaconBlock(node: BeaconNode, blck: BeaconBlock) =
+proc onBeaconBlock*(node: BeaconNode, blck: BeaconBlock) =
   let
     blockRoot = hash_tree_root_final(blck)
     stateSlot = node.beaconState.slot

beacon_chain/sync_protocol.nim

@@ -1,8 +1,8 @@
 import
-  options,
+  options, tables,
   chronicles, eth/[rlp, p2p], chronos, ranges/bitranges, eth/p2p/rlpx,
   spec/[datatypes, crypto, digest],
-  beacon_node, beacon_chain_db, time
+  beacon_node, beacon_chain_db, time, ssz
 
 type
   ValidatorChangeLogEntry* = object
@@ -18,7 +18,30 @@ type
     node*: BeaconNode
     db*: BeaconChainDB
 
-const maxBlocksInRequest = 50
+# const maxBlocksInRequest = 50
+
+func toHeader(b: BeaconBlock): BeaconBlockHeader =
+  discard
+
+proc fromHeader(b: var BeaconBlock, h: BeaconBlockHeader) =
+  discard
+
+proc importBlocks(node: BeaconNode, roots: openarray[(Eth2Digest, uint64)], headers: openarray[BeaconBlockHeader], bodies: openarray[BeaconBlockBody]) =
+  var headerMap = initTable[Eth2Digest, int]()
+  var bodyMap = initTable[Eth2Digest, int]()
+
+  for i, h in headers:
+    headerMap[hash_tree_root_final(h)] = i
+  for i, b in bodies:
+    headerMap[hash_tree_root_final(b)] = i
+
+  for r in roots:
+    let h = r[0]
+    if h in headerMap and h in bodyMap:
+      var blk: BeaconBlock
+      blk.fromHeader(headers[headerMap[h]])
+      blk.body = bodies[bodyMap[h]]
+      node.onBeaconBlock(blk)
 
 p2pProtocol BeaconSync(version = 1,
                        shortName = "bcs",
@@ -28,61 +51,82 @@ p2pProtocol BeaconSync(version = 1,
     const
       protocolVersion = 1 # TODO: Spec doesn't specify this yet
       networkId = 1
+    let node = peer.networkState.node
+
     var
-      latestFinalizedRoot: Eth2Digest
-      latestFinalizedEpoch: uint64
-      bestRoot: Eth2Digest
-      bestSlot: uint64
+      latestFinalizedRoot: Eth2Digest # TODO
+      latestFinalizedEpoch: uint64 = node.beaconState.finalized_epoch
+      bestRoot: Eth2Digest # TODO
+      bestSlot: uint64 = node.beaconState.slot
 
     await peer.status(protocolVersion, networkId, latestFinalizedRoot, latestFinalizedEpoch,
         bestRoot, bestSlot)
 
     let m = await peer.nextMsg(BeaconSync.status)
-
-    if latestFinalizedEpoch == m.latestFinalizedEpoch and bestSlot == m.bestSlot:
-      # ???
+    let bestDiff = cmp((latestFinalizedEpoch, bestSlot), (m.latestFinalizedEpoch, m.bestSlot))
+    if bestDiff == 0:
+      # Nothing to do?
       discard
     else:
-      let betterPeer = if latestFinalizedEpoch == m.latestFinalizedEpoch:
-          bestSlot > m.bestSlot
-        else:
-          latestFinalizedEpoch > m.latestFinalizedEpoch
-
       # TODO: Check for WEAK_SUBJECTIVITY_PERIOD difference and terminate the
       # connection if it's too big.
-
       let db = peer.networkState.db
 
-      if betterPeer:
+      if bestDiff > 0:
         # Send roots
         # TODO: Currently we send all block roots in one "packet". Maybe
         # they should be split to multiple packets.
         type Root = (Eth2Digest, uint64)
         var roots = newSeqOfCap[Root](128)
         for i in m.bestSlot .. bestSlot:
-          discard
-          # for r in db.getBlockRootsForSlot(i):
-          #   roots.add((r, i))
+          for r in db.getBlockRootsForSlot(i):
+            roots.add((r, i))
 
         await peer.beaconBlockRoots(roots)
       else:
         # Receive roots
-        discard
+        let roots = await peer.nextMsg(BeaconSync.beaconBlockRoots)
+        let headers = await peer.getBeaconBlockHeaders(bestRoot, bestSlot, 1024, 0)
+        var bodiesRequest = newSeq[Eth2Digest]()
+        for r in roots.roots:
+          bodiesRequest.add(r[0])
+        let bodies = await peer.getBeaconBlockBodies(bodiesRequest)
+
+        node.importBlocks(roots.roots, headers.get.blockHeaders, bodies.get.blockBodies)
+
 
   proc status(peer: Peer, protocolVersion, networkId: int, latestFinalizedRoot: Eth2Digest,
-        latestFinalizedEpoch: uint64, bestRoot: Eth2Digest, bestSlot: uint64) =
-    discard
+        latestFinalizedEpoch: uint64, bestRoot: Eth2Digest, bestSlot: uint64)
 
   proc beaconBlockRoots(peer: Peer, roots: openarray[(Eth2Digest, uint64)])
 
   requestResponse:
-    proc getBeaconBlockHeaders(peer: Peer, blockRoot: Eth2Digest, slot: uint64, maxHeaders: int, skipSlots: int)
+    proc getBeaconBlockHeaders(peer: Peer, blockRoot: Eth2Digest, slot: uint64, maxHeaders: int, skipSlots: int) =
+      # TODO: validate maxHeaders
+      var s = slot
+      var headers = newSeq[BeaconBlockHeader](maxHeaders)
+      let db = peer.networkState.db
+      while headers.len < maxHeaders:
+        let blkRoots = db.getBlockRootsForSlot(s)
+        for r in blkRoots:
+          headers.add(db.getBlock(r).get().toHeader)
+          if headers.len == maxHeaders: break
+        inc s
+      await peer.beaconBlockHeaders(reqId, headers)
+
     proc beaconBlockHeaders(peer: Peer, blockHeaders: openarray[BeaconBlockHeader])
 
   requestResponse:
-    proc getBeaconBlockBodies(peer: Peer, blockRoots: openarray[Eth2Digest])
-    proc beaconBlockBodies(peer: Peer, blockBodies: openarray[BeaconBlockBody])
+    proc getBeaconBlockBodies(peer: Peer, blockRoots: openarray[Eth2Digest]) =
+      # TODO: Validate blockRoots.len
+      var bodies = newSeq[BeaconBlockBody](blockRoots.len)
+      let db = peer.networkState.db
+      for r in blockRoots:
+        if (let blk = db.getBlock(r); blk.isSome):
+          bodies.add(blk.get().body)
+      await peer.beaconBlockBodies(reqId, bodies)
 
+    proc beaconBlockBodies(peer: Peer, blockBodies: openarray[BeaconBlockBody])
 
 
   requestResponse: