feat: improved dnr sync logic

Signed-off-by: Soumya Ghosh Dastidar <[email protected]>

Author: gdsoumya

cmd/geth/dbcmd.go

@@ -61,6 +61,8 @@ Remove blockchain and state databases`,
 		Category:  "DATABASE COMMANDS",
 		Subcommands: []cli.Command{
 			dbInspectCmd,
+			dbDeleteDNR,
+			dbGetDNR,
 			dbStatCmd,
 			dbCompactCmd,
 			dbGetCmd,
@@ -85,6 +87,25 @@ Remove blockchain and state databases`,
 		Usage:       "Inspect the storage size for each type of data in the database",
 		Description: `This commands iterates the entire database. If the optional 'prefix' and 'start' arguments are provided, then the iteration is limited to the given subset of data.`,
 	}
+	dbDeleteDNR = cli.Command{
+		Action:    utils.MigrateFlags(deleteDNRData),
+		Name:      "delete-dnr",
+		ArgsUsage: "<epoch (optional)>",
+		Flags: utils.GroupFlags([]cli.Flag{
+			utils.SyncModeFlag,
+		}, utils.NetworkFlags, utils.DatabasePathFlags),
+		Usage:       "Deletes all dnr entries from db",
+		Description: `This commands iterates the entire database and deletes all dnr entries`,
+	}
+	dbGetDNR = cli.Command{
+		Action: utils.MigrateFlags(getDNRKeys),
+		Name:   "get-dnr-data",
+		Flags: utils.GroupFlags([]cli.Flag{
+			utils.SyncModeFlag,
+		}, utils.NetworkFlags, utils.DatabasePathFlags),
+		Usage:       "Deletes all dnr entries from db",
+		Description: `This commands iterates the entire database and deletes all dnr entries`,
+	}
 	dbCheckStateContentCmd = cli.Command{
 		Action:    utils.MigrateFlags(checkStateContent),
 		Name:      "check-state-content",
@@ -301,6 +322,66 @@ func inspect(ctx *cli.Context) error {
 	return rawdb.InspectDatabase(db, prefix, start)
 }
 
+func deleteDNRData(ctx *cli.Context) error {
+	var (
+		epoch []byte
+	)
+	if ctx.NArg() > 1 {
+		return fmt.Errorf("Max 1 arguments: %v", ctx.Command.ArgsUsage)
+	}
+	if ctx.NArg() >= 1 {
+		if d, err := hexutil.Decode(ctx.Args().Get(0)); err != nil {
+			return fmt.Errorf("failed to hex-decode 'prefix': %v", err)
+		} else {
+			epoch = d
+		}
+	}
+	stack, _ := makeConfigNode(ctx)
+	defer stack.Close()
+
+	db := utils.MakeChainDatabase(ctx, stack, false)
+	defer db.Close()
+
+	if epoch != nil {
+		key := "dnr-" + string(epoch)
+		if err := db.Delete([]byte(key)); err != nil {
+			log.Error("failed to delete dnr key", "key", string(key))
+			return err
+		}
+	} else {
+		iter := db.NewIterator([]byte("dnr-"), nil)
+		for iter.Next() {
+			key := iter.Key()
+			log.Info("deleting dnr key ", "key", string(key))
+			if err := db.Delete(key); err != nil {
+				log.Error("failed to delete dnr key", "key", string(key))
+				return err
+			}
+		}
+	}
+	log.Info("deleted dnr data")
+	return nil
+}
+
+func getDNRKeys(ctx *cli.Context) error {
+	if ctx.NArg() > 0 {
+		return fmt.Errorf("No arguments: %v", ctx.Command.ArgsUsage)
+	}
+	stack, _ := makeConfigNode(ctx)
+	defer stack.Close()
+
+	db := utils.MakeChainDatabase(ctx, stack, true)
+	defer db.Close()
+
+	iter := db.NewIterator([]byte("dnr-"), nil)
+	for iter.Next() {
+		key := iter.Key()
+		log.Info("dnr key ", "key", string(key))
+	}
+
+	return nil
+}
+
 func checkStateContent(ctx *cli.Context) error {
 	var (
 		prefix []byte

consensus/clique/clique.go

@@ -382,7 +382,7 @@ func (c *Clique) snapshot(chain consensus.ChainHeaderReader, number uint64, hash
 		// If an on-disk checkpoint snapshot can be found, use that
 		if epoch {
 			if s, err := loadSnapshot(c.config, c.signatures, c.db, number); err == nil {
-				log.Warn("Loaded voting snapshot from disk", "block", number, "hash", hash, "snap", snap)
+				log.Warn("Loaded voting snapshot from disk", "block", number, "hash", hash, "snap", s)
 				snap = s
 				break
 			} else {
@@ -642,7 +642,7 @@ func (c *Clique) Seal(chain consensus.ChainHeaderReader, block *types.Block, res
 					log.Warn("failed to update snapshot", "error", err)
 					return
 				}
-				log.Info("epoch stored", "header_no,", header.Number.Uint64(), "hash", header.Hash())
+				log.Info("epoch stored", "header_no", header.Number.Uint64(), "hash", header.Hash())
 			}
 		default:
 			log.Warn("Sealing result is not read by miner", "sealhash", SealHash(header))

consensus/clique/ren.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/big"
+	"os"
+	"strconv"
 	"sync"
 	"time"
 
@@ -17,7 +19,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 )
 
-const BlockBuffer = 100
+const SyncBlockBuffer = 100
 const LogSubnetUpdated = "0x203322e912486658ef4fc95d5da32dcbed12f5fceac274c0b6618c5874bb892f"
 const LogNewEpoch = "0xaf2fc4796f2932ce294c3684deffe5098d3ef65dc2dd64efa80ef94eed88b01e"
 
@@ -53,6 +55,15 @@ func NewDNR(config *params.CliqueConfig, db ethdb.Database) *DNR {
 }
 
 func (d *DNR) Watch(ctx context.Context, db ethdb.Database) {
+	syncWait := int64(30)
+	syncPeriodEnv := os.Getenv("SYNC_WAIT_PERIOD")
+	if syncPeriodEnv != "" {
+		if tempPeriod, err := strconv.ParseInt(syncPeriodEnv, 10, 64); err == nil {
+			log.Info("custom dnr sync period", "period", syncPeriodEnv)
+			syncWait = tempPeriod
+		}
+	}
+
 	d.syncLock.Lock()
 	d.synced = false
 	d.syncLock.Unlock()
@@ -63,6 +74,12 @@ func (d *DNR) Watch(ctx context.Context, db ethdb.Database) {
 	}
 	lastBlock := d.LastEpochBlock
 
+	lastSyncedBlock, err := GetLastSyncedBlock(db)
+
+	if err == nil && lastBlock < lastSyncedBlock {
+		lastBlock = lastSyncedBlock
+	}
+
 	log.Warn("starting watch...", "existing_validators", d.Validators, "last_epoch", d.LastEpochBlock, "dnr_addr", d.config.DNR)
 
 	for {
@@ -144,7 +161,15 @@ func (d *DNR) Watch(ctx context.Context, db ethdb.Database) {
 
 		lastBlock = lastBlockNumber.Uint64()
 
-		time.Sleep(time.Second * 30)
+		if (lastBlock - lastSyncedBlock) >= 2*SyncBlockBuffer {
+			if err := StoreLastSyncedBlock(db, lastBlock-SyncBlockBuffer); err != nil {
+				log.Warn("failed to store last synced dnr block", "new_block", lastBlock, "synced_to", lastSyncedBlock)
+			} else {
+				lastSyncedBlock = lastBlock - SyncBlockBuffer
+			}
+		}
+
+		time.Sleep(time.Second * time.Duration(syncWait))
 	}
 }
 
@@ -197,3 +222,15 @@ func GetDNR(db ethdb.Database, epoch uint64) (*DNR, error) {
 	}
 	return &dnr, nil
 }
+
+func StoreLastSyncedBlock(db ethdb.Database, block uint64) error {
+	return db.Put([]byte("dnr-last-sync"), new(big.Int).SetUint64(block).Bytes())
+}
+
+func GetLastSyncedBlock(db ethdb.Database) (uint64, error) {
+	blob, err := db.Get([]byte("dnr-last-sync"))
+	if err != nil {
+		return 0, err
+	}
+	return new(big.Int).SetBytes(blob).Uint64(), nil
+}

consensus/clique/snapshot.go

@@ -20,13 +20,14 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"sort"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
 	lru "github.com/hashicorp/golang-lru"
-	"sort"
 )
 
 // Vote represents a single vote that an authorized signer made to modify the
@@ -173,10 +174,15 @@ func (s *Snapshot) signers() []common.Address {
 // inturn returns if a signer at a given block height is in-turn or not.
 func (s *Snapshot) inturn(number uint64, signer common.Address) bool {
 	signers, offset := s.signers(), 0
-	for offset < len(signers) && signers[offset] != signer {
+	signerCount := len(signers)
+	if signerCount == 0 {
+		log.Warn("no signers in the validator list, blocks cannot be produced")
+		return false
+	}
+	for offset < signerCount && signers[offset] != signer {
 		offset++
 	}
-	return (number % uint64(len(signers))) == uint64(offset)
+	return (number % uint64(signerCount)) == uint64(offset)
 }
 
 // signers retrieves the list of authorized signers in ascending order.