Moved ETX Set to FIFO PMT with header commitment

consensus/blake3pow/consensus.go

@@ -628,6 +628,7 @@ func (blake3pow *Blake3pow) Finalize(chain consensus.ChainHeaderReader, header *
 	}
 	header.Header().SetUTXORoot(state.UTXORoot())
 	header.Header().SetEVMRoot(state.IntermediateRoot(true))
+	header.Header().SetEtxSetRoot(state.ETXRoot())
 }
 
 // FinalizeAndAssemble implements consensus.Engine, accumulating the block and

consensus/progpow/consensus.go

@@ -677,6 +677,7 @@ func (progpow *Progpow) Finalize(chain consensus.ChainHeaderReader, header *type
 	}
 	header.Header().SetUTXORoot(state.UTXORoot())
 	header.Header().SetEVMRoot(state.IntermediateRoot(true))
+	header.Header().SetEtxSetRoot(state.ETXRoot())
 }
 
 // FinalizeAndAssemble implements consensus.Engine, accumulating the block and

core/block_validator.go

@@ -109,7 +109,7 @@ func (v *BlockValidator) ValidateBody(block *types.WorkObject) error {
 // transition, such as amount of used gas, the receipt roots and the state root
 // itself. ValidateState returns a database batch if the validation was a success
 // otherwise nil and an error is returned.
-func (v *BlockValidator) ValidateState(block *types.WorkObject, statedb *state.StateDB, receipts types.Receipts, utxoEtxs []*types.Transaction, etxSet *types.EtxSet, usedGas uint64) error {
+func (v *BlockValidator) ValidateState(block *types.WorkObject, statedb *state.StateDB, receipts types.Receipts, utxoEtxs []*types.Transaction, usedGas uint64) error {
 	start := time.Now()
 	header := types.CopyHeader(block.Header())
 	time1 := common.PrettyDuration(time.Since(start))
@@ -132,6 +132,9 @@ func (v *BlockValidator) ValidateState(block *types.WorkObject, statedb *state.S
 	if root := statedb.UTXORoot(); header.UTXORoot() != root {
 		return fmt.Errorf("invalid utxo root (remote: %x local: %x)", header.UTXORoot(), root)
 	}
+	if root := statedb.ETXRoot(); header.EtxSetRoot() != root {
+		return fmt.Errorf("invalid etx root (remote: %x local: %x)", header.EtxSetRoot(), root)
+	}
 	time5 := common.PrettyDuration(time.Since(start))
 	// Collect ETXs emitted from each successful transaction
 	var emittedEtxs types.Transactions
@@ -148,19 +151,6 @@ func (v *BlockValidator) ValidateState(block *types.WorkObject, statedb *state.S
 	if etxHash := types.DeriveSha(emittedEtxs, trie.NewStackTrie(nil)); etxHash != header.EtxHash() {
 		return fmt.Errorf("invalid etx hash (remote: %x local: %x)", header.EtxHash(), etxHash)
 	}
-	// Confirm the ETX set used by the block matches the ETX set given in the block body
-	// This is the resulting ETX set after all ETXs in the block have been processed
-	// After validation, this ETX set should be stored in the database
-	if etxSet != nil {
-		etxSetHash := etxSet.Hash()
-		if etxSetHash != block.EtxSetHash() {
-			return fmt.Errorf("expected ETX Set hash %x does not match block ETXSetHash %x", etxSetHash, block.EtxSetHash())
-		}
-	} else {
-		if block.EtxSetHash() != types.EmptyEtxSetHash {
-			return fmt.Errorf("expected ETX Set hash %x does not match block ETXSetHash %x", types.EmptyRootHash, block.EtxSetHash())
-		}
-	}
 
 	// Check that the UncledS in the header matches the S from the block
 	expectedUncledS := v.engine.UncledLogS(block)

core/bodydb.go

@@ -91,7 +91,7 @@ func NewBodyDb(db ethdb.Database, engine consensus.Engine, hc *HeaderChain, chai
 }
 
 // Append
-func (bc *BodyDb) Append(block *types.WorkObject, newInboundEtxs types.Transactions) ([]*types.Log, error) {
+func (bc *BodyDb) Append(block *types.WorkObject) ([]*types.Log, error) {
 	bc.chainmu.Lock()
 	defer bc.chainmu.Unlock()
 
@@ -102,7 +102,7 @@ func (bc *BodyDb) Append(block *types.WorkObject, newInboundEtxs types.Transacti
 	var err error
 	if nodeCtx == common.ZONE_CTX && bc.ProcessingState() {
 		// Process our block
-		logs, err = bc.processor.Apply(batch, block, newInboundEtxs)
+		logs, err = bc.processor.Apply(batch, block)
 		if err != nil {
 			return nil, err
 		}

core/core.go

@@ -1120,8 +1120,8 @@ func (c *Core) State() (*state.StateDB, error) {
 }
 
 // StateAt returns a new mutable state based on a particular point in time.
-func (c *Core) StateAt(root common.Hash, utxoRoot common.Hash) (*state.StateDB, error) {
-	return c.sl.hc.bc.processor.StateAt(root, utxoRoot)
+func (c *Core) StateAt(root, utxoRoot, etxRoot common.Hash) (*state.StateDB, error) {
+	return c.sl.hc.bc.processor.StateAt(root, utxoRoot, etxRoot)
 }
 
 // StateCache returns the caching database underpinning the blockchain instance.

core/genesis.go

@@ -186,7 +186,7 @@ func SetupGenesisBlockWithOverride(db ethdb.Database, genesis *Genesis, nodeLoca
 	// We have the genesis block in database(perhaps in ancient database)
 	// but the corresponding state is missing.
 	header := rawdb.ReadHeader(db, stored, 0)
-	if _, err := state.New(header.EVMRoot(), header.UTXORoot(), state.NewDatabaseWithConfig(db, nil), state.NewDatabaseWithConfig(db, nil), nil, nodeLocation, logger); err != nil {
+	if _, err := state.New(header.EVMRoot(), header.UTXORoot(), header.EtxSetRoot(), state.NewDatabaseWithConfig(db, nil), state.NewDatabaseWithConfig(db, nil), state.NewDatabaseWithConfig(db, nil), nil, nodeLocation, logger); err != nil {
 		if genesis == nil {
 			genesis = DefaultGenesisBlock()
 		}
@@ -286,7 +286,7 @@ func (g *Genesis) ToBlock(startingExpansionNumber uint64) *types.WorkObject {
 	}
 	head.Header().SetCoinbase(common.Zero)
 	head.Header().SetBaseFee(new(big.Int).SetUint64(params.InitialBaseFee))
-	head.Header().SetEtxSetHash(types.EmptyEtxSetHash)
+	head.Header().SetEtxSetRoot(types.EmptyRootHash)
 	if g.GasLimit == 0 {
 		head.Header().SetGasLimit(params.GenesisGasLimit)
 	}
@@ -432,7 +432,7 @@ func DefaultLocalGenesisBlock(consensusEngine string) *Genesis {
 			Nonce:      66,
 			ExtraData:  hexutil.MustDecode("0x11bbe8db4e347b4e8c937c1c8370e4b5ed33adb3db69cbdb7a38e1e50b1b82fb"),
 			GasLimit:   5000000,
-			Difficulty: big.NewInt(100000),
+			Difficulty: big.NewInt(10000),
 		}
 	}
 	return &Genesis{

core/headerchain.go

@@ -341,10 +341,10 @@ func (hc *HeaderChain) setStateProcessing() bool {
 }
 
 // Append
-func (hc *HeaderChain) AppendBlock(block *types.WorkObject, newInboundEtxs types.Transactions) error {
+func (hc *HeaderChain) AppendBlock(block *types.WorkObject) error {
 	blockappend := time.Now()
 	// Append block else revert header append
-	logs, err := hc.bc.Append(block, newInboundEtxs)
+	logs, err := hc.bc.Append(block)
 	if err != nil {
 		return err
 	}
@@ -473,47 +473,29 @@ func (hc *HeaderChain) SetCurrentState(head *types.WorkObject) error {
 		if hc.IsGenesisHash(header.Hash()) {
 			break
 		}
-		// Checking of the Etx set exists makes sure that we have processed the
-		// state of the parent block
-		etxSet := rawdb.ReadEtxSet(hc.headerDb, header.Hash(), header.NumberU64(nodeCtx))
-		if etxSet != nil {
+
+		// Check if the state has been processed for this block
+		processedState := rawdb.ReadProcessedState(hc.headerDb, header.Hash())
+		if processedState {
 			break
 		}
 		current = types.CopyWorkObject(header)
 	}
 
 	// Run through the hash stack to update canonicalHash and forward state processor
 	for i := len(headersWithoutState) - 1; i >= 0; i-- {
-		err := hc.ReadInboundEtxsAndAppendBlock(headersWithoutState[i])
+		block := hc.GetBlockOrCandidate(headersWithoutState[i].Hash(), headersWithoutState[i].NumberU64(nodeCtx))
+		if block == nil {
+			return errors.New("could not find block during SetCurrentState: " + headersWithoutState[i].Hash().String())
+		}
+		err := hc.AppendBlock(block)
 		if err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-// ReadInboundEtxsAndAppendBlock reads the inbound etxs from database and appends the block
-func (hc *HeaderChain) ReadInboundEtxsAndAppendBlock(header *types.WorkObject) error {
-	nodeCtx := hc.NodeCtx()
-	block := hc.GetBlockOrCandidate(header.Hash(), header.NumberU64(nodeCtx))
-	if block == nil {
-		return errors.New("could not find block during reorg")
-	}
-	_, order, err := hc.engine.CalcOrder(block)
-	if err != nil {
-		return err
-	}
-	var inboundEtxs types.Transactions
-	if order < nodeCtx {
-		inboundEtxs = rawdb.ReadInboundEtxs(hc.headerDb, header.Hash())
-	}
-	err = hc.AppendBlock(block, inboundEtxs)
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
 // findCommonAncestor
 func (hc *HeaderChain) findCommonAncestor(header *types.WorkObject) *types.WorkObject {
 	current := types.CopyWorkObject(header)
@@ -1030,8 +1012,8 @@ func (hc *HeaderChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.S
 	return hc.scope.Track(hc.chainSideFeed.Subscribe(ch))
 }
 
-func (hc *HeaderChain) StateAt(root common.Hash, utxoRoot common.Hash) (*state.StateDB, error) {
-	return hc.bc.processor.StateAt(root, utxoRoot)
+func (hc *HeaderChain) StateAt(root, utxoRoot, etxRoot common.Hash) (*state.StateDB, error) {
+	return hc.bc.processor.StateAt(root, utxoRoot, etxRoot)
 }
 
 func (hc *HeaderChain) SlicesRunning() []common.Location {

core/rawdb/accessors_chain.go

@@ -132,6 +132,20 @@ func DeleteHeaderNumber(db ethdb.KeyValueWriter, hash common.Hash) {
 	}
 }
 
+func ReadProcessedState(db ethdb.KeyValueReader, hash common.Hash) bool {
+	data, _ := db.Get(processedStateKey(hash))
+	if len(data) == 0 {
+		return false
+	}
+	return data[0] == 1
+}
+
+func WriteProcessedState(db ethdb.KeyValueWriter, hash common.Hash) {
+	if err := db.Put(processedStateKey(hash), []byte{1}); err != nil {
+		db.Logger().WithField("err", err).Fatal("Failed to store processed state for block " + hash.String())
+	}
+}
+
 // ReadHeadHeaderHash retrieves the hash of the current canonical head header.
 func ReadHeadHeaderHash(db ethdb.KeyValueReader) common.Hash {
 	data, _ := db.Get(headHeaderKey)

core/rawdb/accessors_chain_test.go

@@ -11,12 +11,12 @@
 
 // Tests block header storage and retrieval operations.
 func TestHeaderStorage(t *testing.T) {
 	db := NewMemoryDatabase()
 
 	// Create a test header to move around the database and make sure it's really new
 	header := types.EmptyHeader(2)
 	header.SetParentHash(common.Hash{1}, common.ZONE_CTX)
 	header.SetBaseFee(big.NewInt(1))
 
 	if entry := ReadHeader(db, header.Hash(), common.ZONE_CTX); entry != nil {
 		t.Fatalf("Non existent header returned: %v", entry)
@@ -38,7 +38,7 @@
 
 // Tests termini storage and retrieval operations.
 func TestTerminiStorage(t *testing.T) {
 	db := NewMemoryDatabase()
 
 	// Create a test termini to move around the database and make sure it's really new
 	termini := types.EmptyTermini()
@@ -60,32 +60,9 @@
 	}
 }
 
-func TestEtxSetStorage(t *testing.T) {
-	db := NewMemoryDatabase()
-
-	// Create a test etxSet to move around the database and make sure it's really new
-	etxSet := types.NewEtxSet()
-	hash := common.Hash{1}
-	var number uint64 = 0
-	if entry := ReadEtxSet(db, hash, number); entry != nil {
-		t.Fatalf("Non existent etxSet returned: %v", entry)
-	}
-	t.Log("EtxSet Hash stored", hash)
-	// Write and verify the etxSet in the database
-	WriteEtxSet(db, hash, 0, etxSet)
-	if entry := ReadEtxSet(db, hash, number); entry == nil {
-		t.Fatalf("Stored etxSet not found with hash %s", hash)
-	}
-	// Delete the etxSet and verify the execution
-	DeleteEtxSet(db, hash, number)
-	if entry := ReadEtxSet(db, hash, number); entry != nil {
-		t.Fatalf("Deleted etxSet returned: %v", entry)
-	}
-}
-
 // Tests inbound etx storage and retrieval operations.
 func TestInboundEtxsStorage(t *testing.T) {
 	db := NewMemoryDatabase()
 	hash := common.Hash{1}
 
 	to := common.BytesToAddress([]byte{0x01}, common.Location{0, 0})
@@ -127,14 +104,14 @@
 
 // Tests block header storage and retrieval operations.
 func TestWorkObjectStorage(t *testing.T) {
 	db := NewMemoryDatabase()
 
 	// Create a test header to move around the database and make sure it's really new
 	woBody := &types.WorkObjectBody{}
 	woBody.SetTransactions([]*types.Transaction{})
 	woBody.SetExtTransactions([]*types.Transaction{})
 	woBody.SetHeader(types.EmptyHeader(2).Header())
 	header := types.NewWorkObject(types.NewWorkObjectHeader(types.EmptyRootHash, types.EmptyRootHash, big.NewInt(11), big.NewInt(30000), types.EmptyRootHash, types.BlockNonce{23}, common.LocationFromAddressBytes([]byte{0x01, 0x01})), woBody, nil)
 
 	if entry := ReadWorkObject(db, header.Hash(), types.BlockObject); entry != nil {
 		t.Fatalf("Non existent header returned: %v", entry)

core/rawdb/schema.go

@@ -108,6 +108,7 @@ var (
 	UtxoPrefix                  = []byte("ut")    // outpointPrefix + hash -> types.Outpoint
 	spentUTXOsPrefix            = []byte("sutxo") // spentUTXOsPrefix + hash -> []types.SpentTxOut
 	AddressUtxosPrefix          = []byte("au")    // addressUtxosPrefix + hash -> []types.UtxoEntry
+	processedStatePrefix        = []byte("ps")    // processedStatePrefix + hash -> boolean
 
 	blockBodyPrefix         = []byte("b")   // blockBodyPrefix + num (uint64 big endian) + hash -> block body
 	blockReceiptsPrefix     = []byte("r")   // blockReceiptsPrefix + num (uint64 big endian) + hash -> block receipts
@@ -268,6 +269,10 @@ func headerNumberKey(hash common.Hash) []byte {
 	return append(headerNumberPrefix, hash.Bytes()...)
 }
 
+func processedStateKey(hash common.Hash) []byte {
+	return append(processedStatePrefix, hash.Bytes()...)
+}
+
 // blockBodyKey = blockBodyPrefix + num (uint64 big endian) + hash
 func blockBodyKey(number uint64, hash common.Hash) []byte {
 	return append(append(blockBodyPrefix, encodeBlockNumber(number)...), hash.Bytes()...)

core/slice.go

@@ -343,8 +343,9 @@ func (sl *Slice) Append(header *types.WorkObject, domPendingHeader *types.WorkOb
 		subReorg = sl.miningStrategy(bestPh, tempPendingHeader)
 
 		if order < nodeCtx {
-			// Store the inbound etxs for dom blocks that did not get picked and use
+			// Store the inbound etxs for all dom blocks and use
 			// it in the future if dom switch happens
+			// This should be pruned at the re-org tolerance depth
 			rawdb.WriteInboundEtxs(sl.sliceDb, block.Hash(), newInboundEtxs)
 		}
 
@@ -458,6 +459,7 @@ func (sl *Slice) Append(header *types.WorkObject, domPendingHeader *types.WorkOb
 		"gasLimit":     block.GasLimit(),
 		"evmRoot":      block.EVMRoot(),
 		"utxoRoot":     block.UTXORoot(),
+		"etxSetRoot":   block.EtxSetRoot(),
 		"order":        order,
 		"location":     block.Location(),
 		"elapsed":      common.PrettyDuration(time.Since(start)),
@@ -1254,7 +1256,6 @@ func (sl *Slice) init() error {
 		if err != nil {
 			return err
 		}
-		rawdb.WriteEtxSet(sl.sliceDb, genesisHash, 0, types.NewEtxSet())
 		// This is just done for the startup process
 		sl.hc.SetCurrentHeader(genesisHeader)
 
@@ -1376,10 +1377,11 @@ func (sl *Slice) combinePendingHeader(header *types.WorkObject, slPendingHeader
 		combinedPendingHeader.Header().SetUncleHash(header.UncleHash())
 		combinedPendingHeader.Header().SetTxHash(header.Header().TxHash())
 		combinedPendingHeader.Header().SetEtxHash(header.EtxHash())
-		combinedPendingHeader.Header().SetEtxSetHash(header.EtxSetHash())
+		combinedPendingHeader.Header().SetEtxSetRoot(header.EtxSetRoot())
 		combinedPendingHeader.Header().SetReceiptHash(header.ReceiptHash())
 		combinedPendingHeader.Header().SetEVMRoot(header.EVMRoot())
 		combinedPendingHeader.Header().SetUTXORoot(header.UTXORoot())
+		combinedPendingHeader.Header().SetEtxSetRoot(header.EtxSetRoot())
 		combinedPendingHeader.Header().SetCoinbase(header.Coinbase())
 		combinedPendingHeader.Header().SetBaseFee(header.BaseFee())
 		combinedPendingHeader.Header().SetGasLimit(header.GasLimit())
@@ -1437,7 +1439,6 @@ func (sl *Slice) WriteGenesisBlock(block *types.WorkObject, location common.Loca
 	sl.AddPendingEtxsRollup(types.PendingEtxsRollup{block, emptyPendingEtxs})
 	sl.hc.AddBloom(types.Bloom{}, block.Hash())
 	sl.hc.currentHeader.Store(block)
-	rawdb.WriteEtxSet(sl.sliceDb, block.Hash(), block.NumberU64(sl.NodeCtx()), types.NewEtxSet())
 }
 
 // NewGenesisPendingHeader creates a pending header on the genesis block