feat: add etcd compact

metadium/admin.go

@@ -102,6 +102,8 @@ var (
 	ErrNotRunning     = errors.New("not running")
 	ErrAlreadyRunning = errors.New("already running")
 	ErrInvalidEnode   = errors.New("invalid enode")
+
+	etcdCompactFrequency = int64(100)
 )
 
 func (n *metaNode) eq(m *metaNode) bool {
@@ -1084,12 +1086,22 @@ func LogBlock(height int64, hash common.Hash) {
 	}
 
 	tstart := time.Now()
-	if err := admin.etcdPut("metadium-work", string(work)); err != nil {
+	rev, err := admin.etcdPut("metadium-work", string(work))
+	if err != nil {
 		log.Error("Metadium - failed to log the latest block",
 			"height", height, "hash", hash, "took", time.Since(tstart))
 	} else {
 		log.Info("Metadium - logged the latest block",
 			"height", height, "hash", hash, "took", time.Since(tstart))
+
+		if (rev%etcdCompactFrequency == 0) && (rev > 100) {
+			go func() {
+				if err := admin.etcdCompact(rev); err != nil {
+					log.Error("Metadium - failed to compact",
+						"rev", rev, "took", time.Since(tstart))
+				}
+			}()
+		}
 	}
 
 	admin.blocksMined++

metadium/etcdutil.go

@@ -14,6 +14,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/coreos/etcd/clientv3"
 	"github.com/coreos/etcd/embed"
 	"github.com/coreos/etcd/etcdserver/api/membership"
 	"github.com/coreos/etcd/etcdserver/api/v3client"
@@ -442,16 +443,16 @@ func (ma *metaAdmin) etcdLeader(locked bool) (uint64, *metaNode) {
 	return 0, nil
 }
 
-func (ma *metaAdmin) etcdPut(key, value string) error {
+func (ma *metaAdmin) etcdPut(key, value string) (int64, error) {
 	if !ma.etcdIsRunning() {
-		return ErrNotRunning
+		return 0, ErrNotRunning
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(),
 		ma.etcd.Server.Cfg.ReqTimeout())
 	defer cancel()
-	_, err := ma.etcdCli.Put(ctx, key, value)
-	return err
+	resp, err := ma.etcdCli.Put(ctx, key, value)
+	return resp.Header.Revision, err
 }
 
 func (ma *metaAdmin) etcdGet(key string) (string, error) {
@@ -487,6 +488,20 @@ func (ma *metaAdmin) etcdDelete(key string) error {
 	return err
 }
 
+func (ma *metaAdmin) etcdCompact(rev int64) error {
+	if !ma.etcdIsRunning() {
+		return ErrNotRunning
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(),
+		ma.etcd.Server.Cfg.ReqTimeout())
+	defer cancel()
+	_, err := ma.etcdCli.Compact(ctx, rev, clientv3.WithCompactPhysical())
+	// WithCompactPhysical makes Compact wait until all compacted entries are
+	// removed from the etcd server's storage.
+	return err
+}
+
 func (ma *metaAdmin) etcdInfo() interface{} {
 	if ma.etcd == nil {
 		return ErrNotRunning