bound block quarantine size (#1564)

* bound block quarantine size

* add additional logging for block quarantining

* re-add quarantine.add() call

* remove pre-finalization blocks; add logging for full quarantine

* clear quarantine on chain reorganization

* update block_sim and tests

* update test_attestation_pool

beacon_chain/block_pools/chain_dag.nim

@@ -17,7 +17,7 @@ import
   ../spec/[
     crypto, datatypes, digest, helpers, validator, state_transition,
     beaconstate],
-  block_pools_types
+  block_pools_types, quarantine
 
 export block_pools_types
 
@@ -691,7 +691,8 @@ proc delState(dag: ChainDAGRef, bs: BlockSlot) =
     dag.db.delState(root.get())
     dag.db.delStateRoot(bs.blck.root, bs.slot)
 
-proc updateHead*(dag: ChainDAGRef, newHead: BlockRef) =
+proc updateHead*(
+    dag: ChainDAGRef, newHead: BlockRef, quarantine: var QuarantineRef) =
   ## Update what we consider to be the current head, as given by the fork
   ## choice.
   ## The choice of head affects the choice of finalization point - the order
@@ -738,6 +739,7 @@ proc updateHead*(dag: ChainDAGRef, newHead: BlockRef) =
       finalized = shortLog(dag.headState.data.data.finalized_checkpoint)
 
     # A reasonable criterion for "reorganizations of the chain"
+    quarantine.clearQuarantine()
     beacon_reorgs_total.inc()
   else:
     info "Updated head block",

beacon_chain/block_pools/clearance.nim

@@ -225,7 +225,8 @@ proc addRawBlock*(
   # the pending dag calls this function back later in a loop, so as long
   # as dag.add(...) requires a SignedBeaconBlock, easier to keep them in
   # pending too.
-  quarantine.add(dag, signedBlock)
+  if not quarantine.add(dag, signedBlock):
+    debug "Block quarantine full"
 
   # TODO possibly, it makes sense to check the database - that would allow sync
   #      to simply fill up the database with random blocks the other clients
@@ -343,8 +344,10 @@ proc isValidBeaconBlock*(
     # ChainDAGRef.add(...) directly, with no additional validity checks. TODO,
     # not specific to this, but by the pending dag keying on the htr of the
     # BeaconBlock, not SignedBeaconBlock, opens up certain spoofing attacks.
-    debug "parent unknown, putting block in quarantine"
-    quarantine.add(dag, signed_beacon_block)
+    debug "parent unknown, putting block in quarantine",
+      current_slot = shortLog(current_slot)
+    if not quarantine.add(dag, signed_beacon_block):
+      debug "Block quarantine full"
     return err(MissingParent)
 
   # [REJECT] The current finalized_checkpoint is an ancestor of block -- i.e.

beacon_chain/block_pools/quarantine.nim

@@ -46,10 +46,47 @@ func addMissing*(quarantine: var QuarantineRef, root: Eth2Digest) =
     # If the block is in orphans, we no longer need it
     discard quarantine.missing.hasKeyOrPut(root, MissingBlock())
 
+func removeOldBlocks(quarantine: var QuarantineRef, dag: ChainDAGRef) =
+  var oldBlocks: seq[Eth2Digest]
+
+  for k, v in quarantine.orphans.pairs():
+    if v.message.slot <= dag.finalizedHead.slot:
+      oldBlocks.add k
+
+  for k in oldBlocks:
+    quarantine.orphans.del k
+
+func clearQuarantine*(quarantine: var QuarantineRef) =
+  quarantine.orphans.clear()
+  quarantine.missing.clear()
+
 func add*(quarantine: var QuarantineRef, dag: ChainDAGRef,
-          signedBlock: SignedBeaconBlock) =
+          signedBlock: SignedBeaconBlock): bool =
   ## Adds block to quarantine's `orphans` and `missing` lists.
+
+  # Typically, blocks will arrive in mostly topological order, with some
+  # out-of-order block pairs. Therefore, it is unhelpful to use either a
+  # FIFO or LIFO discpline, and since by definition each block gets used
+  # either 0 or 1 times it's not a cache either. Instead, stop accepting
+  # new blocks, and rely on syncing to cache up again if necessary. When
+  # using forward sync, blocks only arrive in an order not requiring the
+  # quarantine.
+  #
+  # For typical use cases, this need not be large, as they're two or three
+  # blocks arriving out of order due to variable network delays. As blocks
+  # for future slots are rejected before reaching quarantine, this usually
+  # will be a block for the last couple of slots for which the parent is a
+  # likely imminent arrival.
+  const MAX_QUARANTINE_ORPHANS = 10
+
+  quarantine.removeOldBlocks(dag)
+
+  if quarantine.orphans.len >= MAX_QUARANTINE_ORPHANS:
+    return false
+
   quarantine.orphans[signedBlock.root] = signedBlock
   quarantine.missing.del(signedBlock.root)
 
   quarantine.addMissing(signedBlock.message.parent_root)
+
+  true

beacon_chain/eth2_processor.nim

@@ -72,7 +72,7 @@ proc updateHead*(self: var Eth2Processor, wallSlot: Slot): BlockRef =
   # justified and finalized
   let oldFinalized = self.chainDag.finalizedHead.blck
 
-  self.chainDag.updateHead(newHead)
+  self.chainDag.updateHead(newHead, self.quarantine)
   beacon_head_root.set newHead.root.toGaugeValue
 
   # Cleanup the fork choice v2 if we have a finalized head

research/block_sim.nim

@@ -142,7 +142,7 @@ cli do(slots = SLOTS_PER_EPOCH * 6,
         attPool.addForkChoice(epochRef, blckRef, signedBlock.message, blckRef.slot)
 
       blck() = added[]
-      chainDag.updateHead(added[])
+      chainDag.updateHead(added[], quarantine)
 
   for i in 0..<slots:
     let

tests/test_attestation_pool.nim

@@ -348,7 +348,7 @@ suiteReport "Attestation pool processing" & preset():
 
         let head = pool[].selectHead(blockRef[].slot)
         doassert: head == blockRef[]
-        chainDag.updateHead(head)
+        chainDag.updateHead(head, quarantine)
 
         attestations.setlen(0)
         for index in 0'u64 ..< committees_per_slot:
@@ -417,7 +417,7 @@ suiteReport "Attestation validation " & preset():
         pool[].addForkChoice(epochRef, blckRef, signedBlock.message, blckRef.slot)
 
       check: added.isOk()
-      chainDag.updateHead(added[])
+      chainDag.updateHead(added[], quarantine)
 
     var
       # Create an attestation for slot 1!

tests/test_block_pool.nim

@@ -173,7 +173,7 @@ suiteReport "Block pool processing" & preset():
     check:
       b4Add[].parent == b2Add[]
 
-    dag.updateHead(b4Add[])
+    dag.updateHead(b4Add[], quarantine)
 
     var blocks: array[3, BlockRef]
 
@@ -225,7 +225,7 @@ suiteReport "Block pool processing" & preset():
 
       b2Get.get().refs.parent == b1Get.get().refs
 
-    dag.updateHead(b2Get.get().refs)
+    dag.updateHead(b2Get.get().refs, quarantine)
 
     # The heads structure should have been updated to contain only the new
     # b2 head
@@ -258,7 +258,7 @@ suiteReport "Block pool processing" & preset():
     let
       b1Add = dag.addRawBlock(quarantine, b1, nil)
 
-    dag.updateHead(b1Add[])
+    dag.updateHead(b1Add[], quarantine)
 
     check:
       dag.head == b1Add[]
@@ -349,7 +349,7 @@ suiteReport "chain DAG finalization tests" & preset():
           tmpState[].data, dag.head.root, tmpState[].data.slot, cache, {}))
       let added = dag.addRawBlock(quarantine, blck, nil)
       check: added.isOk()
-      dag.updateHead(added[])
+      dag.updateHead(added[], quarantine)
 
     check:
       dag.heads.len() == 1
@@ -407,7 +407,7 @@ suiteReport "chain DAG finalization tests" & preset():
         dag.headState.data, dag.head.root, cache)
       let added = dag.addRawBlock(quarantine, blck, nil)
       check: added.isOk()
-      dag.updateHead(added[])
+      dag.updateHead(added[], quarantine)
 
     check:
       dag.heads.len() == 1
@@ -443,7 +443,7 @@ suiteReport "chain DAG finalization tests" & preset():
         true):
       let added = dag.addRawBlock(quarantine, blck, nil)
       check: added.isOk()
-      dag.updateHead(added[])
+      dag.updateHead(added[], quarantine)
 
     # Advance past epoch so that the epoch transition is gapped
     check:
@@ -458,7 +458,7 @@ suiteReport "chain DAG finalization tests" & preset():
 
     let added = dag.addRawBlock(quarantine, blck, nil)
     check: added.isOk()
-    dag.updateHead(added[])
+    dag.updateHead(added[], quarantine)
 
     let
       dag2 = init(ChainDAGRef, defaultRuntimePreset, db)