[staking] patch for staking protocol

action/protocol/staking/builder.go

@@ -8,5 +8,6 @@ type (
 	BuilderConfig struct {
 		Staking              genesis.Staking
 		PersistCandsMapBlock uint64
+		CandsMapPatchDir     string
 	}
 )

action/protocol/staking/candidate_test.go

@@ -58,6 +58,15 @@ func TestSer(t *testing.T) {
 	l1 := &CandidateList{}
 	r.NoError(l1.Deserialize(ser))
 	r.Equal(l, l1)
+
+	// empty CandidateList can successfully Serialize/Deserialize
+	var m CandidateList
+	ser, err = m.Serialize()
+	r.NoError(err)
+	r.Equal([]byte{}, ser)
+	var m1 CandidateList
+	r.NoError(m1.Deserialize(ser))
+	r.Nil(m1)
 }
 
 func TestClone(t *testing.T) {

action/protocol/staking/patchstore.go

@@ -0,0 +1,80 @@
+// Copyright (c) 2020 IoTeX Foundation
+// This is an alpha (internal) release and is not suitable for production. This source code is provided 'as is' and no
+// warranties are given as to title or non-infringement, merchantability or fitness for purpose and, to the extent
+// permitted by law, all liability for your use of the code is disclaimed. This source code is governed by Apache
+// License 2.0 that can be found in the LICENSE file.
+
+package staking
+
+import (
+	"encoding/csv"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+
+	"github.com/pkg/errors"
+)
+
+const (
+	_name     = "name"
+	_operator = "operator"
+)
+
+// PatchStore is the patch store of staking protocol
+type PatchStore struct {
+	dir string
+}
+
+// NewPatchStore creates a new staking patch store
+func NewPatchStore(dir string) *PatchStore {
+	return &PatchStore{dir: dir}
+}
+
+func (store *PatchStore) pathOf(height uint64) string {
+	return filepath.Join(store.dir, fmt.Sprintf("%d.patch", height))
+}
+
+func (store *PatchStore) read(reader *csv.Reader) (CandidateList, error) {
+	record, err := reader.Read()
+	if err != nil {
+		return nil, err
+	}
+	if len(record) != 1 {
+		return nil, errors.Errorf("invalid record %+v", record)
+	}
+	data, err := hex.DecodeString(record[0])
+	if err != nil {
+		return nil, err
+	}
+	var list CandidateList
+	if err := list.Deserialize(data); err != nil {
+		return nil, err
+	}
+	return list, nil
+}
+
+// Read reads CandidateList by name and CandidateList by operator of given height
+func (store *PatchStore) Read(height uint64) (CandidateList, CandidateList, CandidateList, error) {
+	file, err := os.Open(store.pathOf(height))
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	reader := csv.NewReader(file)
+	reader.FieldsPerRecord = -1
+	listByName, err := store.read(reader)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	listByOperator, err := store.read(reader)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	listByOwner, err := store.read(reader)
+	if err != nil && err != io.EOF {
+		// io.EOF indicates an empty owner list
+		return nil, nil, nil, err
+	}
+	return listByName, listByOperator, listByOwner, nil
+}

action/protocol/staking/patchstore_test.go

@@ -0,0 +1,38 @@
+// Copyright (c) 2019 IoTeX Foundation
+// This is an alpha (internal) release and is not suitable for production. This source code is provided 'as is' and no
+// warranties are given as to title or non-infringement, merchantability or fitness for purpose and, to the extent
+// permitted by law, all liability for your use of the code is disclaimed. This source code is governed by Apache
+// License 2.0 that can be found in the LICENSE file.
+
+package staking
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestInvalidDirectory(t *testing.T) {
+	require := require.New(t)
+	dir := filepath.Join(t.TempDir(), "invalid")
+	_, err := os.Create(dir)
+	require.NoError(err)
+	_, _, _, err = NewPatchStore(dir).Read(0)
+	require.ErrorContains(err, "not a directory")
+}
+
+func TestInvalidDirectory2(t *testing.T) {
+	require := require.New(t)
+	dir := t.TempDir()
+	require.NoError(os.Remove(dir))
+	_, err := os.Stat(dir)
+	require.ErrorIs(err, os.ErrNotExist)
+	_, _, _, err = NewPatchStore(dir).Read(0)
+	require.ErrorContains(err, "no such file or directory")
+}
+
+func TestCorruptedData(t *testing.T) {
+	// TODO: add test for corrupted data
+}

action/protocol/staking/protocol.go

@@ -80,6 +80,7 @@ type (
 		config             Configuration
 		candBucketsIndexer *CandidatesBucketsIndexer
 		voteReviser        *VoteReviser
+		patch              *PatchStore
 	}
 
 	// Configuration is the staking protocol configuration.
@@ -90,6 +91,7 @@ type (
 		MinStakeAmount        *big.Int
 		BootstrapCandidates   []genesis.BootstrapCandidate
 		PersistCandsMapBlock  uint64
+		CandsMapPatchDir      string
 	}
 
 	// DepositGas deposits gas to some pool
@@ -150,6 +152,7 @@ func NewProtocol(depositGas DepositGas, cfg *BuilderConfig, candBucketsIndexer *
 			MinStakeAmount:        minStakeAmount,
 			BootstrapCandidates:   cfg.Staking.BootstrapCandidates,
 			PersistCandsMapBlock:  cfg.PersistCandsMapBlock,
+			CandsMapPatchDir:      cfg.CandsMapPatchDir,
 		},
 		depositGas:         depositGas,
 		candBucketsIndexer: candBucketsIndexer,
@@ -179,7 +182,10 @@ func (p *Protocol) Start(ctx context.Context, sr protocol.StateReader) (interfac
 	if p.needToReadCandsMap(height) {
 		name, operator, owners, err := readCandCenterStateFromStateDB(sr, height)
 		if err != nil {
-			return nil, errors.Wrap(err, "failed to read name/operator map")
+			// stateDB does not have name/operator map yet
+			if name, operator, owners, err = p.readCandCenterStateFromPatch(height); err != nil {
+				return nil, errors.Wrap(err, "failed to read name/operator map")
+			}
 		}
 		if err = c.candCenter.base.loadNameOperatorMapOwnerList(name, operator, owners); err != nil {
 			return nil, errors.Wrap(err, "failed to load name/operator map to cand center")
@@ -336,10 +342,13 @@ func (p *Protocol) PreCommit(ctx context.Context, sm protocol.StateManager) erro
 	name := base.candsInNameMap()
 	op := base.candsInOperatorMap()
 	owners := base.ownersList()
-	if len(name) == 0 || len(op) == 0 || len(owners) == 0 {
+	if len(name) == 0 || len(op) == 0 {
 		return ErrNilParameters
 	}
-	return errors.Wrap(p.writeCandCenterStateToStateDB(sm, name, op, owners), "failed to write name/operator map to stateDB")
+	if err := p.writeCandCenterStateToStateDB(sm, name, op, owners); err != nil {
+		return errors.Wrap(err, "failed to write name/operator map to stateDB")
+	}
+	return nil
 }
 
 // Commit commits the last change
@@ -627,3 +636,10 @@ func (p *Protocol) writeCandCenterStateToStateDB(sm protocol.StateManager, name,
 	_, err := sm.PutState(owners, protocol.NamespaceOption(CandsMapNS), protocol.KeyOption(_ownerKey))
 	return err
 }
+
+func (p *Protocol) readCandCenterStateFromPatch(height uint64) (CandidateList, CandidateList, CandidateList, error) {
+	if p.patch == nil {
+		p.patch = NewPatchStore(p.config.CandsMapPatchDir)
+	}
+	return p.patch.Read(height)
+}

blockchain/config.go

@@ -28,6 +28,7 @@ type (
 		ChainDBPath            string           `yaml:"chainDBPath"`
 		TrieDBPatchFile        string           `yaml:"trieDBPatchFile"`
 		TrieDBPath             string           `yaml:"trieDBPath"`
+		CandsMapPatchDir       string           `yaml:"candsMapPatchDir"`
 		IndexDBPath            string           `yaml:"indexDBPath"`
 		BloomfilterIndexDBPath string           `yaml:"bloomfilterIndexDBPath"`
 		CandidateIndexDBPath   string           `yaml:"candidateIndexDBPath"`
@@ -78,6 +79,7 @@ var (
 		ChainDBPath:            "/var/data/chain.db",
 		TrieDBPatchFile:        "/var/data/trie.db.patch",
 		TrieDBPath:             "/var/data/trie.db",
+		CandsMapPatchDir:       "/var/data",
 		IndexDBPath:            "/var/data/index.db",
 		BloomfilterIndexDBPath: "/var/data/bloomfilter.index.db",
 		CandidateIndexDBPath:   "/var/data/candidate.index.db",

chainservice/builder.go

@@ -451,6 +451,7 @@ func (builder *Builder) registerStakingProtocol() error {
 		&staking.BuilderConfig{
 			Staking:              builder.cfg.Genesis.Staking,
 			PersistCandsMapBlock: builder.cfg.Chain.PersistCandsMapBlock,
+			CandsMapPatchDir:     builder.cfg.Chain.CandsMapPatchDir,
 		},
 		builder.cs.candBucketsIndexer,
 		builder.cfg.Genesis.GreenlandBlockHeight,

db/db_bolt_test.go

@@ -74,6 +74,9 @@ func TestBucketExists(t *testing.T) {
 	r.False(kv.BucketExists("name"))
 	r.NoError(kv.Put("name", []byte("key"), []byte{}))
 	r.True(kv.BucketExists("name"))
+	v, err := kv.Get("name", []byte("key"))
+	r.NoError(err)
+	r.Equal([]byte{}, v)
 }
 
 func BenchmarkBoltDB_Get(b *testing.B) {