Create database tables for block / blob data sidecar quarantine

Add low-level interface to persist blocks by block root and sidecars
by block root + index. Also support deleting junk that's been finalized
and deletion of all data associated with a block root.

Can pass `db.getQuarantineDB()` into the quarantine to expose the new DB
tables while hiding APIs that deal with trusted data, and then use the
new `beacon_chain_db_quarantine.nim` APIs to store / fetch / delete.

Author: etan-status

beacon_chain/beacon_chain_db.nim

@@ -20,7 +20,10 @@ import
           forks,
           presets,
           state_transition],
-  "."/[beacon_chain_db_light_client, filepath]
+  "."/[beacon_chain_db_light_client,
+       beacon_chain_db_quarantine,
+       db_utils,
+       filepath]
 
 from ./spec/datatypes/capella import BeaconState
 from ./spec/datatypes/deneb import TrustedSignedBeaconBlock
@@ -153,6 +156,10 @@ type
       ##
       ## See `summaries` for an index in the other direction.
 
+    quarantine: QuarantineDB
+      ## Pending data that passed basic checks including proposer signature
+      ## but that is not fully validated / trusted yet.
+
     lcData: LightClientDataDB
       ## Persistent light client data to avoid expensive recomputations
 
@@ -593,6 +600,36 @@ proc new*(T: type BeaconChainDB,
   if cfg.FULU_FORK_EPOCH != FAR_FUTURE_EPOCH:
     columns = kvStore db.openKvStore("fulu_columns").expectDb()
 
+  let quarantine = db.initQuarantineDB(QuarantineDBNames(
+    # TODO Set these to non-empty string to actually use them
+    # once we have a new quarantine implemented. If empty string is used,
+    # no database table is created!
+    phase0Blocks:
+      "",  # "blocks_quarantine"
+    altairBlocks:
+      "",  # "altair_blocks_quarantine"
+    bellatrixBlocks:
+      "",  # "bellatrix_blocks_quarantine"
+    capellaBlocks:
+      "",  # "capella_blocks_quarantine"
+    denebBlocks:
+      "",  # "deneb_blocks_quarantine"
+    electraBlocks:
+      "",  # "electra_blocks_quarantine"
+    fuluBlocks:
+      if cfg.FULU_FORK_EPOCH != FAR_FUTURE_EPOCH:
+        ""  # "fulu_blocks_quarantine"
+      else:
+        "",
+    denebDataSidecars:
+      "",  # "deneb_blobs_quarantine"
+    fuluDataSidecars:
+      if cfg.FULU_FORK_EPOCH != FAR_FUTURE_EPOCH:
+        ""  # "fulu_columns_quarantine"
+      else:
+        "")).expectDb()
+  static: doAssert ConsensusFork.high == ConsensusFork.Fulu
+
   # Versions prior to 1.4.0 (altair) stored validators in `immutable_validators`
   # which stores validator keys in compressed format - this is
   # slow to load and has been superceded by `immutable_validators2` which uses
@@ -634,6 +671,7 @@ proc new*(T: type BeaconChainDB,
     stateDiffs: stateDiffs,
     summaries: summaries,
     finalizedBlocks: finalizedBlocks,
+    quarantine: quarantine,
     lcData: lcData
   )
 
@@ -657,6 +695,9 @@ proc new*(T: type BeaconChainDB,
         dir, "nbc", readOnly = readOnly, manualCheckpoint = true).expectDb()
   BeaconChainDB.new(db, cfg)
 
+template getQuarantineDB*(db: BeaconChainDB): QuarantineDB =
+  db.quarantine
+
 template getLightClientDataDB*(db: BeaconChainDB): LightClientDataDB =
   db.lcData
 
@@ -683,18 +724,6 @@ proc decodeSnappySSZ[T](data: openArray[byte], output: var T): bool =
       err = e.msg, typ = name(T), dataLen = data.len
     false
 
-proc decodeSZSSZ[T](data: openArray[byte], output: var T): bool =
-  try:
-    let decompressed = decodeFramed(data, checkIntegrity = false)
-    readSszBytes(decompressed, output, updateRoot = false)
-    true
-  except CatchableError as e:
-    # If the data can't be deserialized, it could be because it's from a
-    # version of the software that uses a different SSZ encoding
-    warn "Unable to deserialize data, old database?",
-      err = e.msg, typ = name(T), dataLen = data.len
-    false
-
 func encodeSSZ*(v: auto): seq[byte] =
   try:
     SSZ.encode(v)
@@ -708,14 +737,6 @@ func encodeSnappySSZ(v: auto): seq[byte] =
     # In-memory encode shouldn't fail!
     raiseAssert err.msg
 
-func encodeSZSSZ(v: auto): seq[byte] =
-  # https://github.com/google/snappy/blob/main/framing_format.txt
-  try:
-    encodeFramed(SSZ.encode(v))
-  except CatchableError as err:
-    # In-memory encode shouldn't fail!
-    raiseAssert err.msg
-
 proc getRaw(db: KvStoreRef, key: openArray[byte], T: type Eth2Digest): Opt[T] =
   var res: Opt[T]
   proc decode(data: openArray[byte]) =
@@ -796,6 +817,7 @@ proc close*(db: BeaconChainDB) =
   if db.db == nil: return
 
   # Close things roughly in reverse order
+  db.quarantine.close()
   if not isNil(db.columns):
     discard db.columns.close()
   if not isNil(db.blobs):
@@ -1131,7 +1153,7 @@ proc getBlobSidecar*(db: BeaconChainDB, root: Eth2Digest, index: BlobIndex,
                      value: var BlobSidecar): bool =
   db.blobs.getSZSSZ(blobkey(root, index), value) == GetResult.found
 
-proc getDataColumnSidecarSZ*(db: BeaconChainDB, root: Eth2Digest, 
+proc getDataColumnSidecarSZ*(db: BeaconChainDB, root: Eth2Digest,
                              index: ColumnIndex, data: var seq[byte]): bool =
   let dataPtr = addr data # Short-lived
   func decode(data: openArray[byte]) =

beacon_chain/beacon_chain_db_light_client.nim

@@ -15,7 +15,7 @@ import
   # Beacon chain internals
   spec/datatypes/altair,
   spec/[eth2_ssz_serialization, helpers],
-  ./db_limits
+  ./db_utils
 
 logScope: topics = "lcdata"
 
@@ -172,11 +172,6 @@ type
       ## Tracks the finalized sync committee periods for which complete data
       ## has been imported (from `dag.tail.slot`).
 
-template disposeSafe(s: untyped): untyped =
-  if distinctBase(s) != nil:
-    s.dispose()
-    s = typeof(s)(nil)
-
 proc initHeadersStore(
     backend: SqStoreRef,
     name, typeName: string): KvResult[LightClientHeaderStore] =