Merge branch 'master' into cleanup_blockdao

action/protocol/staking/bucket_validation.go

@@ -83,7 +83,7 @@ func validateBucketEndorsement(esm *EndorsementStateManager, bucket *VoteBucket,
 		status := endorse.Status(height)
 		if isEndorsed && status == EndorseExpired {
 			return &handleError{
-				err:           errors.New("bucket is not endorsed"),
+				err:           errors.New("endorse bucket is expired"),
 				failureStatus: iotextypes.ReceiptStatus_ErrInvalidBucketType,
 			}
 		}
@@ -96,7 +96,7 @@ func validateBucketEndorsement(esm *EndorsementStateManager, bucket *VoteBucket,
 	case errors.Is(err, state.ErrStateNotExist):
 		if isEndorsed {
 			return &handleError{
-				err:           errors.New("bucket is not endorsed"),
+				err:           errors.New("bucket is not an endorse bucket"),
 				failureStatus: iotextypes.ReceiptStatus_ErrInvalidBucketType,
 			}
 		}

action/protocol/staking/bucket_validation_test.go

@@ -114,7 +114,7 @@ func TestValidateBucket(t *testing.T) {
 		blkHeight := uint64(10)
 		// not endorsed bucket
 		r.Nil(validateBucketEndorsement(esm, bkt, false, blkHeight))
-		r.ErrorContains(validateBucketEndorsement(esm, bkt, true, blkHeight), "bucket is not endorsed")
+		r.ErrorContains(validateBucketEndorsement(esm, bkt, true, blkHeight), "bucket is not an endorse bucket")
 		// endorsed bucket
 		r.NoError(esm.Put(bktIdx, &Endorsement{ExpireHeight: endorsementNotExpireHeight}))
 		r.Nil(validateBucketEndorsement(esm, bkt, true, blkHeight))
@@ -126,6 +126,6 @@ func TestValidateBucket(t *testing.T) {
 		// endorse expired bucket
 		r.NoError(esm.Put(bktIdx, &Endorsement{ExpireHeight: blkHeight}))
 		r.Nil(validateBucketEndorsement(esm, bkt, false, blkHeight))
-		r.ErrorContains(validateBucketEndorsement(esm, bkt, true, blkHeight), "bucket is not endorsed")
+		r.ErrorContains(validateBucketEndorsement(esm, bkt, true, blkHeight), "endorse bucket is expired")
 	})
 }

action/protocol/staking/handler_candidate_endorsement_test.go

@@ -258,6 +258,33 @@ func TestProtocol_HandleCandidateEndorsement(t *testing.T) {
 			[]expectCandidate{},
 			nil,
 		},
+		{
+			"once endorsed, bucket cannot be unstaked",
+			[]uint64{0, 10},
+			[]uint64{0, 1},
+			1300000,
+			identityset.Address(1),
+			1,
+			uint64(1000000),
+			uint64(1000000),
+			big.NewInt(1000),
+			1,
+			true,
+			true,
+			&appendAction{
+				func() action.Action {
+					act, err := action.NewUnstake(0, 1, []byte{}, uint64(1000000), big.NewInt(1000))
+					require.NoError(err)
+					return act
+				},
+				iotextypes.ReceiptStatus_ErrInvalidBucketType,
+				nil,
+			},
+			nil,
+			iotextypes.ReceiptStatus_Success,
+			[]expectCandidate{},
+			nil,
+		},
 		{
 			"unendorse a valid bucket",
 			[]uint64{0, 9},

action/protocol/staking/handlers.go

@@ -172,6 +172,10 @@ func (p *Protocol) handleUnstake(ctx context.Context, act *action.Unstake, csm C
 			failureStatus: iotextypes.ReceiptStatus_ErrUnstakeBeforeMaturity,
 		}
 	}
+	if rErr := validateBucketEndorsement(NewEndorsementStateManager(csm.SM()), bucket, false, blkCtx.BlockHeight); rErr != nil {
+		return log, rErr
+	}
+	// TODO: cannot unstake if selected as candidates in this or next epoch
 
 	// update bucket
 	bucket.UnstakeStartTime = blkCtx.BlockTimeStamp.UTC()

db/db_bolt.go

@@ -11,6 +11,7 @@ import (
 	"syscall"
 
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	bolt "go.etcd.io/bbolt"
 	"go.uber.org/zap"
 
@@ -25,8 +26,17 @@ const _fileMode = 0600
 var (
 	// ErrDBNotStarted represents the error when a db has not started
 	ErrDBNotStarted = errors.New("db has not started")
+
+	boltdbMtc = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "iotex_boltdb_metrics",
+		Help: "boltdb metrics.",
+	}, []string{"type", "method"})
 )
 
+func init() {
+	prometheus.MustRegister(boltdbMtc)
+}
+
 // BoltDB is KVStore implementation based bolt DB
 type BoltDB struct {
 	lifecycle.Readiness
@@ -305,13 +315,35 @@ func (b *BoltDB) WriteBatch(kvsb batch.KVStoreBatch) (err error) {
 	kvsb.Lock()
 	defer kvsb.Unlock()
 
+	type doubleKey struct {
+		ns  string
+		key string
+	}
+	// remove duplicate keys, only keep the last write for each key
+	entryKeySet := make(map[doubleKey]struct{})
+	uniqEntries := make([]*batch.WriteInfo, 0)
+	for i := kvsb.Size() - 1; i >= 0; i-- {
+		write, e := kvsb.Entry(i)
+		if e != nil {
+			return e
+		}
+		// only handle Put and Delete
+		if write.WriteType() != batch.Put && write.WriteType() != batch.Delete {
+			continue
+		}
+		k := doubleKey{ns: write.Namespace(), key: string(write.Key())}
+		if _, ok := entryKeySet[k]; !ok {
+			entryKeySet[k] = struct{}{}
+			uniqEntries = append(uniqEntries, write)
+		}
+	}
+	boltdbMtc.WithLabelValues(b.path, "entrySize").Set(float64(kvsb.Size()))
+	boltdbMtc.WithLabelValues(b.path, "uniqueEntrySize").Set(float64(len(entryKeySet)))
 	for c := uint8(0); c < b.config.NumRetries; c++ {
 		if err = b.db.Update(func(tx *bolt.Tx) error {
-			for i := 0; i < kvsb.Size(); i++ {
-				write, e := kvsb.Entry(i)
-				if e != nil {
-					return e
-				}
+			// keep order of the writes same as the original batch
+			for i := len(uniqEntries) - 1; i >= 0; i-- {
+				write := uniqEntries[i]
 				ns := write.Namespace()
 				switch write.WriteType() {
 				case batch.Put: