Implement sharding

Author: stwiname

SUBQL.md

@@ -7,11 +7,7 @@
 Many other networks and L2s use forks of geth we can apply the changes to these forks using the following steps
 
 1. Generate a patch between the feature branch and master
-<<<<<<< HEAD
 `git format-patch --stdout <sha1>..<sha2> > subql.patch`
-=======
-`git format-patch --stdout master > subql.patch`
->>>>>>> db90ac101 (Implement transactions bloom filter and subql apis)
 
 2. In the for repo ensure that the subquery geth for is a remote
 
@@ -31,10 +27,8 @@ This will apply the changes but may require resolving some conflicts
 These forks are generally behind the master geth branch. This can mean that further changes are requied to get builds suceeding.
 
 It's suggested to create a patch or patches of these changes to make it easier to sync changes from the fork and from geth in the future
-<<<<<<< HEAD
 
 You can do this by repeating steps 1, 3
 
 e.g `git format-patch --stdout 2d772be398d851a62be53d1b0c162c45bb4876e3..c9760f18c15b8d36af7ef1a9bf5b21056f633b71 > build_fixes.patch`
-=======
->>>>>>> db90ac101 (Implement transactions bloom filter and subql apis)
+

cmd/geth/main.go

@@ -154,6 +154,8 @@ var (
 		utils.BeaconGenesisRootFlag,
 		utils.BeaconGenesisTimeFlag,
 		utils.BeaconCheckpointFlag,
+		utils.ShardStartFlag,
+		utils.ShardEndFlag,
 	}, utils.NetworkFlags, utils.DatabaseFlags)
 
 	rpcFlags = []cli.Flag{

cmd/utils/flags.go

@@ -928,6 +928,17 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
 		Value:    metrics.DefaultConfig.InfluxDBOrganization,
 		Category: flags.MetricsCategory,
 	}
+
+	ShardStartFlag = &cli.Uint64Flag{
+		Name: "shardstart",
+		Usage: "ShardStart the block height of Shard start",
+		Category: flags.StateCategory,
+	}
+	ShardEndFlag = &cli.Uint64Flag{
+		Name: "shardend",
+		Usage: "ShardEnd the block height of Shard end (0 means no end)",
+		Category: flags.StateCategory,
+	}
 )
 
 var (
@@ -1677,6 +1688,25 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 			cfg.EthDiscoveryURLs = SplitAndTrim(urls)
 		}
 	}
+
+	if ctx.IsSet(ShardStartFlag.Name) {
+		start := ctx.Uint64(ShardStartFlag.Name)
+		cfg.ShardStart = &start
+	}
+	if cfg.ShardStart != nil {
+		log.Info("Set shard start", "ShardStart", *cfg.ShardStart)
+	} else {
+		log.Info("Shard start disabled")
+	}
+	if ctx.IsSet(ShardEndFlag.Name) {
+		end := ctx.Uint64(ShardEndFlag.Name)
+		cfg.ShardEnd = &end
+	}
+	if cfg.ShardEnd != nil {
+		log.Info("Set shard end", "shardend", *cfg.ShardEnd)
+	} else {
+		log.Info("Shard end disabled")
+	}
 	// Override any default configs for hard coded networks.
 	switch {
 	case ctx.Bool(MainnetFlag.Name):

common/math/integer.go

@@ -91,3 +91,17 @@ func SafeMul(x, y uint64) (uint64, bool) {
 	hi, lo := bits.Mul64(x, y)
 	return lo, hi != 0
 }
+
+func Min(x int, y int) int {
+	if x <= y {
+		return x
+	}
+	return y
+}
+
+func Max(x int, y int) int {
+	if x >= y {
+		return x
+	}
+	return y
+}

core/blockchain.go

@@ -62,6 +62,9 @@ var (
 	headFinalizedBlockGauge = metrics.NewRegisteredGauge("chain/head/finalized", nil)
 	headSafeBlockGauge      = metrics.NewRegisteredGauge("chain/head/safe", nil)
 
+	tailBlockGauge  = metrics.NewRegisteredGauge("chain/tail/block", nil)
+	tailHeaderGauge = metrics.NewRegisteredGauge("chain/tail/header", nil)
+
 	chainInfoGauge = metrics.NewRegisteredGaugeInfo("chain/info", nil)
 
 	accountReadTimer   = metrics.NewRegisteredResettingTimer("chain/account/reads", nil)
@@ -248,6 +251,7 @@ type BlockChain struct {
 	currentSnapBlock  atomic.Pointer[types.Header] // Current head of snap-sync
 	currentFinalBlock atomic.Pointer[types.Header] // Latest (consensus) finalized block
 	currentSafeBlock  atomic.Pointer[types.Header] // Latest (consensus) safe block
+	earliestBlock     atomic.Pointer[types.Header] // Earliest head of the chain
 
 	bodyCache     *lru.Cache[common.Hash, *types.Body]
 	bodyRLPCache  *lru.Cache[common.Hash, rlp.RawValue]
@@ -524,10 +528,24 @@ func (bc *BlockChain) loadLastState() error {
 			headHeader = header
 		}
 	}
+
+	tailNumber := uint64(0)
+	if dataConfig := rawdb.ReadChainDataConfig(bc.db); dataConfig != nil && dataConfig.DesiredChainDataStart != nil {
+		tailNumber = *dataConfig.DesiredChainDataStart
+	}
+	tailBlock := bc.GetBlockByNumber(tailNumber)
+	if tailBlock == nil {
+		// Corrupt or empty database, init from scratch
+		log.Warn("Tail block missing, resetting chain", "number", tailNumber)
+		return bc.Reset()
+	}
+
 	bc.hc.SetCurrentHeader(headHeader)
+	bc.hc.SetEarliestHeader(tailBlock.Header())
 
 	// Restore the last known head snap block
 	bc.currentSnapBlock.Store(headBlock.Header())
+	bc.earliestBlock.Store(tailBlock.Header())
 	headFastBlockGauge.Update(int64(headBlock.NumberU64()))
 
 	if head := rawdb.ReadHeadFastBlockHash(bc.db); head != (common.Hash{}) {
@@ -900,6 +918,7 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
 			// removed in the hc.SetHead function.
 			rawdb.DeleteBody(db, hash, num)
 			rawdb.DeleteReceipts(db, hash, num)
+			rawdb.DeleteTxBloom(db, hash, num)
 		}
 		// Todo(rjl493456442) txlookup, bloombits, etc
 	}
@@ -936,6 +955,8 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
 		log.Error("SetHead invalidated finalized block")
 		bc.SetFinalized(nil)
 	}
+
+	log.Info("Rewinding blockchain complete", "target", head)
 	return rootNumber, bc.loadLastState()
 }
 
@@ -1152,7 +1173,10 @@ func (bc *BlockChain) Stop() {
 				}
 			}
 			for !bc.triegc.Empty() {
-				triedb.Dereference(bc.triegc.PopItem())
+				// @sq-change
+				root, number := bc.triegc.Pop()
+				triedb.Dereference(root)
+				triedb.SetTail(uint64(-number))
 			}
 			if _, nodes, _ := triedb.Size(); nodes != 0 { // all memory is contained within the nodes return for hashdb
 				log.Error("Dangling trie nodes after full cleanup")
@@ -1526,6 +1550,7 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
 			break
 		}
 		bc.triedb.Dereference(root)
+		bc.triedb.SetTail(uint64(-number))
 	}
 	return nil
 }
@@ -1716,6 +1741,12 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness
 			log.Debug("Abort during block processing")
 			break
 		}
+
+		desiredChainEnd := rawdb.ReadChainDataConfig(bc.db).DesiredChainDataEnd
+		if desiredChainEnd != nil && block.Number().Uint64() > *desiredChainEnd {
+			log.Info("Block after data end, skipping")
+			continue
+		}
 		// If the block is known (in the middle of the chain), it's a special case for
 		// Clique blocks where they can share state among each other, so importing an
 		// older block might complete the state of the subsequent one. In this case,
@@ -2521,3 +2552,113 @@ func (bc *BlockChain) SetTrieFlushInterval(interval time.Duration) {
 func (bc *BlockChain) GetTrieFlushInterval() time.Duration {
 	return time.Duration(bc.flushInterval.Load())
 }
+
+// ----- @sq changed
+func (bc *BlockChain) truncateStateTail(newTail uint64) error {
+	currentTail := bc.triedb.GetTail()
+	if newTail <= currentTail {
+		return nil
+	}
+	for height := currentTail; height < newTail; height++ {
+		header := bc.GetHeaderByNumber(height)
+		// This isn't right,.height cant be used as a priority
+		bc.triegc.Push(header.Root, -int64(height))
+	}
+	// TODO: need to enforce trie truncate on local disk
+	return nil
+}
+
+func (bc *BlockChain) SetTail(height uint64) error {
+	delFn := func(db ethdb.KeyValueWriter, hash common.Hash, num uint64) {
+		frozen, _ := bc.db.Ancients()
+
+		if num+1 <= frozen {
+			// Headerchain should clear out all relevant ancients data
+			log.Crit("delFn called on frozen data")
+		} else {
+			// Remove body and receipts from the active store, remainder will be cleared in the hc.SetTail
+			rawdb.DeleteBody(db, hash, num)
+			rawdb.DeleteReceipts(db, hash, num)
+			rawdb.DeleteTxBloom(db, hash, num)
+		}
+		// Todo(stwiname) from equivalent in SetHead.delFn
+		// Todo(rjl493456442) txlookup, bloombits, etc
+	}
+
+	if err := bc.hc.SetTail(height, delFn); err != nil {
+		return err
+	}
+
+	// TODO delete relevant state
+	// bc.triedb.TruncateTail(num)
+
+	// Clear out any stale content from the caches
+	bc.bodyCache.Purge()
+	bc.bodyRLPCache.Purge()
+	bc.receiptsCache.Purge()
+	bc.blockCache.Purge()
+	bc.txLookupCache.Purge()
+
+	return nil
+}
+
+func (bc *BlockChain) SetShardStartHeight(height uint64) error {
+	config := rawdb.ReadChainDataConfig(bc.db)
+
+	if config.DesiredChainDataEnd != nil && *config.DesiredChainDataEnd <= height {
+		return fmt.Errorf("Start height cannot be after end height")
+	}
+
+	if config.DesiredChainDataStart != nil && *config.DesiredChainDataStart > height {
+		return fmt.Errorf("Start height cannot be decreased, there is no way to recover state")
+	}
+
+	if err := bc.SetTail(height); err != nil {
+		return err
+	}
+
+	// TODO perisisting this should happen with batch in SetTail
+	config.DesiredChainDataStart = &height
+	rawdb.WriteChainDataConfig(bc.db, config)
+
+	return nil
+}
+
+func (bc *BlockChain) SetShardEndHeight(height *uint64) error {
+	config := rawdb.ReadChainDataConfig(bc.db)
+
+	if config.DesiredChainDataStart != nil && *config.DesiredChainDataStart >= *height {
+		return fmt.Errorf("End height cannot be before start height")
+	}
+
+	if height != nil && bc.CurrentHeader().Number.Uint64() > *height {
+		// rollback
+		if err := bc.SetHead(*height); err != nil {
+			return err
+		}
+
+		// TODO truncate state
+
+
+		// type freezer interface {
+		// 	Freeze(threshold uint64) error
+		// }
+		// err = api.eth.chainDb.(freezer).Freeze(0)
+		// if err != nil {
+		// 	return false, err
+		// }
+		// TODO truncate state head
+		// err = api.eth.blockchain.TrieDB().SetHead(*height)
+		// if err != nil {
+		// 	return false, err
+		// }
+
+		// TODO call blockchain.Stop()?
+	}
+	// else blockchain.go will limit this once the desired height is reached
+
+	config.DesiredChainDataEnd = height
+	rawdb.WriteChainDataConfig(bc.db, config)
+
+	return nil
+}

core/blockchain_reader.go

@@ -62,6 +62,10 @@ func (bc *BlockChain) CurrentSafeBlock() *types.Header {
 	return bc.currentSafeBlock.Load()
 }
 
+func (bc *BlockChain) EarliestBlock() *types.Header {
+	return bc.earliestBlock.Load()
+}
+
 // HasHeader checks if a block header is present in the database or not, caching
 // it if present.
 func (bc *BlockChain) HasHeader(hash common.Hash, number uint64) bool {