Prevent database corruption during stateroot computation

execution_chain/core/chain/forked_chain.nim

@@ -482,14 +482,17 @@ proc validateBlock(c: ForkedChainRef,
     parentTxFrame=cast[uint](parentFrame),
     txFrame=cast[uint](txFrame)
 
+  # Update the snapshot before processing the block so that any vertexes in snapshots
+  # from lower levels than the baseTxFrame are removed from the snapshot before running
+  # the stateroot computation.
+  c.updateSnapshot(blk, txFrame)
+
   var receipts = c.processBlock(parent, txFrame, blk, blkHash, finalized).valueOr:
     txFrame.dispose()
     return err(error)
 
   c.writeBaggage(blk, blkHash, txFrame, receipts)
 
-  c.updateSnapshot(blk, txFrame)
-
   let newBlock = c.appendBlock(parent, blk, blkHash, txFrame, move(receipts))
 
   for i, tx in blk.transactions:

execution_chain/core/chain/forked_chain/chain_serialize.nim

@@ -126,12 +126,16 @@ proc replayBlock(fc: ForkedChainRef;
     parentFrame = parent.txFrame
     txFrame = parentFrame.txFrameBegin
 
+  # Update the snapshot before processing the block so that any vertexes in snapshots
+  # from lower levels than the baseTxFrame are removed from the snapshot before running
+  # the stateroot computation.
+  fc.updateSnapshot(blk.blk, txFrame)
+
   var receipts = fc.processBlock(parent, txFrame, blk.blk, blk.hash, false).valueOr:
     txFrame.dispose()
     return err(error)
 
   fc.writeBaggage(blk.blk, blk.hash, txFrame, receipts)
-  fc.updateSnapshot(blk.blk, txFrame)
 
   blk.txFrame = txFrame
   blk.receipts = move(receipts)
@@ -233,11 +237,11 @@ proc deserialize*(fc: ForkedChainRef): Result[void, string] =
 
   # Sanity Checks for the FC state
   if state.latest > state.numBlocks or
-     state.base > state.numBlocks: 
+     state.base > state.numBlocks:
     warn "TODO: Inconsistent state found"
     fc.reset(prevBase)
     return err("Invalid state: latest block is greater than number of blocks")
-  
+
   # Sanity Checks for all the heads in FC state
   for head in state.heads:
     if head > state.numBlocks:

execution_chain/db/aristo/aristo_compute.nim

@@ -74,7 +74,10 @@ proc putKeyAtLevel(
   ## set (vertex data may have been committed to disk without computing the
   ## corresponding hash!)
 
-  if level < db.db.baseTxFrame().level:
+  if level >= db.db.baseTxFrame().level:
+    let txRef = db.deltaAtLevel(level)
+    txRef.layersPutKey(rvid, vtx, key)
+  elif level == dbLevel:
     ?batch.putVtx(db.db, rvid, vtx, key)
 
     if batch.count mod batchSize == 0:
@@ -84,8 +87,11 @@ proc putKeyAtLevel(
         info "Writing computeKey cache", keys = batch.count, accounts = batch.progress
       else:
         debug "Writing computeKey cache", keys = batch.count, accounts = batch.progress
-  else:
-    db.deltaAtLevel(level).layersPutKey(rvid, vtx, key)
+  else: # level > dbLevel but less than baseTxFrame level
+    # Throw defect here because we should not be writing vertexes to the database if
+    # from a lower level than the baseTxFrame level.
+    raiseAssert("Cannot write keys at level < baseTxFrame level. Found level = " &
+        $level & ", baseTxFrame level = " & $db.db.baseTxFrame().level)
 
   ok()
 

execution_chain/db/aristo/aristo_get.nim

@@ -91,6 +91,7 @@ proc getKeyRc*(
       # There was no vertex on the cache. So there must be one the backend (the
       # reason for the key label to exists, at all.)
       return err(GetKeyNotFound)
+
     if vtx[0].isValid:
       return ok ((VOID_HASH_KEY, vtx[0]), vtx[1])
     else:

tests/test_aristo/test_tx_frame.nim

@@ -227,3 +227,46 @@ suite "Aristo TxFrame":
     db.persist(batch, tx2)
     check:
       db.putEndFn(batch).isOk()
+
+  test "Get state root on a txFrame which has lower level than the baseTxFrame":
+    # level 1
+    let tx1 = db.txFrameBegin(db.baseTxFrame())
+    for i in 1..<100:
+      let acc = makeAccount(i.uint64)
+      check tx1.mergeAccountRecord(acc[0], acc[1]).isOk()
+    tx1.checkpoint(1, skipSnapshot = false)
+
+    # level 2
+    let tx2 = db.txFrameBegin(tx1)
+    for i in 100..<200:
+      let acc = makeAccount(i.uint64)
+      check tx2.mergeAccountRecord(acc[0], acc[1]).isOk()
+    tx2.checkpoint(2, skipSnapshot = false)
+
+    # level 3
+    let tx3 = db.txFrameBegin(tx2)
+    for i in 200..<300:
+      let acc = makeAccount(i.uint64)
+      check tx3.mergeAccountRecord(acc[0], acc[1]).isOk()
+    tx3.checkpoint(3, skipSnapshot = false)
+
+    # level 2
+    let tx4 = db.txFrameBegin(tx1)
+    for i in 300..<400:
+      let acc = makeAccount(i.uint64)
+      check tx4.mergeAccountRecord(acc[0], acc[1]).isOk()
+    tx4.checkpoint(2, skipSnapshot = false)
+
+    block:
+      let batch = db.putBegFn().expect("working batch")
+      db.persist(batch, tx3) # after this the baseTxFrame is at level 3
+      check:
+        db.putEndFn(batch).isOk()
+
+    # Verify that getting the state root of the level 3 txFrame does not impact
+    # the persisted state in the database.
+    let stateRootBefore = tx3.fetchStateRoot().get()
+    expect(Defect):
+      discard tx4.fetchStateRoot()
+    let stateRootAfter = tx3.fetchStateRoot().get()
+    check stateRootBefore == stateRootAfter