History pruning (fixes #4419) (#4445)

Introduce (optional) pruning of historical data - a pruned node will
continue to answer queries for historical data up to
`MIN_EPOCHS_FOR_BLOCK_REQUESTS` epochs, or roughly 5 months, capping
typical database usage at around 60-70gb.

To enable pruning, add `--history=prune` to the command line - on the
first start, old data will be cleared (which may take a while) - after
that, data is pruned continuously.

When pruning an existing database, the database will not shrink -
instead, the freed space is recycled as the node continues to run - to
free up space, perform a trusted node sync with a fresh database.

When switching on archive mode in a pruned node, history is retained
from that point onwards.

History pruning is scheduled to be enabled by default in a future
release.

In this PR, `minimal` mode from #4419 is not implemented meaning
retention periods for states and blocks are always the same - depending
on user demand, a future PR may implement `minimal` as well.

AllTests-mainnet.md

@@ -374,6 +374,11 @@ OK: 1/1 Fail: 0/1 Skip: 0/1
 + deletePeer() test                                                                          OK
 ```
 OK: 12/12 Fail: 0/12 Skip: 0/12
+## Pruning
+```diff
++ prune states                                                                               OK
+```
+OK: 1/1 Fail: 0/1 Skip: 0/1
 ## Remove keystore testing suite
 ```diff
 + vesion 1                                                                                   OK
@@ -615,4 +620,4 @@ OK: 2/2 Fail: 0/2 Skip: 0/2
 OK: 9/9 Fail: 0/9 Skip: 0/9
 
 ---TOTAL---
-OK: 344/349 Fail: 0/349 Skip: 5/349
+OK: 345/350 Fail: 0/350 Skip: 5/350

beacon_chain/beacon_chain_db.nim

@@ -70,7 +70,7 @@ type
     ##
     ## 1.2 moved BeaconStateNoImmutableValidators to a separate table to
     ## alleviate some of the btree balancing issues - this doubled the speed but
-    ## was still
+    ## was still slow
     ##
     ## 1.3 creates `kvstore` with rowid, making it quite fast, but doesn't do
     ## anything about existing databases. Versions after that use a separate
@@ -420,41 +420,43 @@ proc loadImmutableValidators(vals: DbSeq[ImmutableValidatorDataDb2]): seq[Immuta
       withdrawal_credentials: tmp.withdrawal_credentials)
 
 template withManyWrites*(dbParam: BeaconChainDB, body: untyped) =
-  let db = dbParam
-  # Make sure we're not nesting transactions.
-  if isInsideTransaction(db.db):
-    raiseAssert "Sqlite does not support nested transactions"
+  let
+    db = dbParam
+    nested = isInsideTransaction(db.db)
+
   # We don't enforce strong ordering or atomicity requirements in the beacon
   # chain db in general, relying instead on readers to be able to deal with
   # minor inconsistencies - however, putting writes in a transaction is orders
   # of magnitude faster when doing many small writes, so we use this as an
   # optimization technique and the templace is named accordingly.
-  expectDb db.db.exec("BEGIN TRANSACTION;")
+  if not nested:
+    expectDb db.db.exec("BEGIN TRANSACTION;")
   var commit = false
   try:
-    body
-    commit = true
+      body
+      commit = true
   finally:
-    if commit:
-      expectDb db.db.exec("COMMIT TRANSACTION;")
-    else:
-      # https://www.sqlite.org/lang_transaction.html
-      #
-      # For all of these errors, SQLite attempts to undo just the one statement
-      # it was working on and leave changes from prior statements within the same
-      # transaction intact and continue with the transaction. However, depending
-      # on the statement being evaluated and the point at which the error occurs,
-      # it might be necessary for SQLite to rollback and cancel the entire transaction.
-      # An application can tell which course of action SQLite took by using the
-      # sqlite3_get_autocommit() C-language interface.
-      #
-      # It is recommended that applications respond to the errors listed above by
-      # explicitly issuing a ROLLBACK command. If the transaction has already been
-      # rolled back automatically by the error response, then the ROLLBACK command
-      # will fail with an error, but no harm is caused by this.
-      #
-      if isInsideTransaction(db.db): # calls `sqlite3_get_autocommit`
-        expectDb db.db.exec("ROLLBACK TRANSACTION;")
+    if not nested:
+      if commit:
+        expectDb db.db.exec("COMMIT TRANSACTION;")
+      else:
+        # https://www.sqlite.org/lang_transaction.html
+        #
+        # For all of these errors, SQLite attempts to undo just the one statement
+        # it was working on and leave changes from prior statements within the same
+        # transaction intact and continue with the transaction. However, depending
+        # on the statement being evaluated and the point at which the error occurs,
+        # it might be necessary for SQLite to rollback and cancel the entire transaction.
+        # An application can tell which course of action SQLite took by using the
+        # sqlite3_get_autocommit() C-language interface.
+        #
+        # It is recommended that applications respond to the errors listed above by
+        # explicitly issuing a ROLLBACK command. If the transaction has already been
+        # rolled back automatically by the error response, then the ROLLBACK command
+        # will fail with an error, but no harm is caused by this.
+        #
+        if isInsideTransaction(db.db): # calls `sqlite3_get_autocommit`
+          expectDb db.db.exec("ROLLBACK TRANSACTION;")
 
 proc new*(T: type BeaconChainDBV0,
           db: SqStoreRef,
@@ -839,29 +841,34 @@ proc putStateRoot*(db: BeaconChainDB, root: Eth2Digest, slot: Slot,
 proc putStateDiff*(db: BeaconChainDB, root: Eth2Digest, value: BeaconStateDiff) =
   db.stateDiffs.putSnappySSZ(root.data, value)
 
-proc delBlock*(db: BeaconChainDB, key: Eth2Digest) =
+proc delBlock*(db: BeaconChainDB, fork: BeaconBlockFork, key: Eth2Digest): bool =
+  var deleted = false
   db.withManyWrites:
-    for kv in db.blocks:
-      kv.del(key.data).expectDb()
-    db.summaries.del(key.data).expectDb()
+    discard db.summaries.del(key.data).expectDb()
+    deleted = db.blocks[fork].del(key.data).expectDb()
+  deleted
 
-proc delState*(db: BeaconChainDB, key: Eth2Digest) =
-  db.withManyWrites:
-    for kv in db.statesNoVal:
-      kv.del(key.data).expectDb()
+proc delState*(db: BeaconChainDB, fork: BeaconStateFork, key: Eth2Digest) =
+  discard db.statesNoVal[fork].del(key.data).expectDb()
+
+proc clearBlocks*(db: BeaconChainDB, fork: BeaconBlockFork) =
+  discard db.blocks[fork].clear().expectDb()
+
+proc clearStates*(db: BeaconChainDB, fork: BeaconStateFork) =
+  discard db.statesNoVal[fork].clear().expectDb()
 
 proc delKeyValue*(db: BeaconChainDB, key: array[1, byte]) =
-  db.keyValues.del(key).expectDb()
-  db.v0.backend.del(key).expectDb()
+  discard db.keyValues.del(key).expectDb()
+  discard db.v0.backend.del(key).expectDb()
 
 proc delKeyValue*(db: BeaconChainDB, key: DbKeyKind) =
   db.delKeyValue(subkey(key))
 
 proc delStateRoot*(db: BeaconChainDB, root: Eth2Digest, slot: Slot) =
-  db.stateRoots.del(stateRootKey(root, slot)).expectDb()
+  discard db.stateRoots.del(stateRootKey(root, slot)).expectDb()
 
 proc delStateDiff*(db: BeaconChainDB, root: Eth2Digest) =
-  db.stateDiffs.del(root.data).expectDb()
+  discard db.stateDiffs.del(root.data).expectDb()
 
 proc putHeadBlock*(db: BeaconChainDB, key: Eth2Digest) =
   db.keyValues.putRaw(subkey(kHeadBlock), key)
@@ -1067,7 +1074,6 @@ proc getBlockSZ*(
   of BeaconBlockFork.EIP4844:
     getBlockSZ(db, key, data, eip4844.TrustedSignedBeaconBlock)
 
-
 proc getStateOnlyMutableValidators(
     immutableValidators: openArray[ImmutableValidatorData2],
     store: KvStoreRef, key: openArray[byte],
@@ -1338,6 +1344,12 @@ proc containsState*(db: BeaconChainDBV0, key: Eth2Digest): bool =
     db.backend.contains(sk).expectDb() or
     db.backend.contains(subkey(phase0.BeaconState, key)).expectDb()
 
+proc containsState*(db: BeaconChainDB, fork: BeaconStateFork, key: Eth2Digest,
+    legacy: bool = true): bool =
+  if db.statesNoVal[fork].contains(key.data).expectDb(): return true
+
+  (legacy and fork == BeaconStateFork.Phase0 and db.v0.containsState(key))
+
 proc containsState*(db: BeaconChainDB, key: Eth2Digest, legacy: bool = true): bool =
   for fork in countdown(BeaconStateFork.high, BeaconStateFork.low):
     if db.statesNoVal[fork].contains(key.data).expectDb(): return true
@@ -1418,7 +1430,7 @@ iterator getAncestorSummaries*(db: BeaconChainDB, root: Eth2Digest):
       INNER JOIN next ON `key` == substr(v, 9, 32)
   )
   SELECT v FROM next;
-"""
+  """
   let
     stmt = expectDb db.db.prepareStmt(
       summariesQuery, array[32, byte],

beacon_chain/conf.nim

@@ -107,6 +107,10 @@ type
     Json = "json"
     None = "none"
 
+  HistoryMode* {.pure.} = enum
+    Archive = "archive"
+    Prune = "prune"
+
   SlashProtCmd* = enum
     `import` = "Import a EIP-3076 slashing protection interchange file"
     `export` = "Export a EIP-3076 slashing protection interchange file"
@@ -576,6 +580,11 @@ type
         defaultValue: ""
         name: "payload-builder-url" .}: string
 
+      historyMode* {.
+        desc: "Retention strategy for historical data (archive/pruned)"
+        defaultValue: HistoryMode.Archive
+        name: "history".}: HistoryMode
+
     of BNStartUpCmd.createTestnet:
       testnetDepositsFile* {.
         desc: "A LaunchPad deposits file for the genesis state validators"

beacon_chain/consensus_object_pools/block_pools_types.nim

@@ -363,10 +363,12 @@ template frontfill*(dagParam: ChainDAGRef): Opt[BlockId] =
     dag.genesis
 
 func horizon*(dag: ChainDAGRef): Slot =
-  ## The sync horizon that we target during backfill - ie we will not backfill
-  ## blocks older than this from the network
-  if dag.head.slot.epoch > dag.cfg.MIN_EPOCHS_FOR_BLOCK_REQUESTS:
-    start_slot(dag.head.slot.epoch - dag.cfg.MIN_EPOCHS_FOR_BLOCK_REQUESTS)
+  ## The sync horizon that we target during backfill - we will backfill and
+  ## retain this and newer blocks, but anything older may get pruned depending
+  ## on the history mode
+  let minSlots = dag.cfg.MIN_EPOCHS_FOR_BLOCK_REQUESTS * SLOTS_PER_EPOCH
+  if dag.head.slot > minSlots:
+    min(dag.finalizedHead.slot, dag.head.slot - minSlots)
   else:
     GENESIS_SLOT