Repository: matrixorigin/matrixone
Pull request #6012 (2022-10-27 06:53:54)

[feat] #6011 Support truncation in logservice.
See the github issue page more details.

## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #6011 

## What this PR does / why we need it:

Support truncation in logservice.

fengttt: @mergifyio refresh

mergify[bot]: > refresh

#### ✅ Pull request refreshed



<!---
DO NOT EDIT
-*- Mergify Payload -*-
{"command": "refresh", "conclusion": "success"}
-*- Mergify Payload End -*-
-->

Patch:
diff --git a/go.mod b/go.mod
index cd3c862e2..d08293650 100644
--- a/go.mod
+++ b/go.mod
@@ -126,4 +126,4 @@ require (
 // see https://github.com/hashicorp/memberlist/pull/273 for progress
 replace github.com/hashicorp/memberlist => github.com/lni/memberlist v0.3.2-0.20220811084252-8a89d1ea99cb
 
-replace github.com/lni/dragonboat/v4 v4.0.0-20220815145555-6f622e8bcbef => github.com/matrixorigin/dragonboat/v4 v4.0.0-20220815145555-6f622e8bcbef
+replace github.com/lni/dragonboat/v4 v4.0.0-20220815145555-6f622e8bcbef => github.com/matrixorigin/dragonboat/v4 v4.0.0-20221026081827-193f8e8b5bb6
diff --git a/go.sum b/go.sum
index aa47e8557..d39cbe5f2 100644
--- a/go.sum
+++ b/go.sum
@@ -312,8 +312,8 @@ github.com/lni/vfs v0.2.1-0.20220616104132-8852fd867376/go.mod h1:LOatfyR8Xeej1j
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
-github.com/matrixorigin/dragonboat/v4 v4.0.0-20220815145555-6f622e8bcbef h1:Udh5GWnPGXDcVWwyKLsWlfoaumF8xVMJmQD1d5L0T0g=
-github.com/matrixorigin/dragonboat/v4 v4.0.0-20220815145555-6f622e8bcbef/go.mod h1:bR6ZGoUwApH4/P4+AezNGT0xehljg+j+Z/Q2Fz+Y4a0=
+github.com/matrixorigin/dragonboat/v4 v4.0.0-20221026081827-193f8e8b5bb6 h1:9BFbf231mh1+I+3srcMncbISTr3cSUxMoAlF/vyCVmI=
+github.com/matrixorigin/dragonboat/v4 v4.0.0-20221026081827-193f8e8b5bb6/go.mod h1:bR6ZGoUwApH4/P4+AezNGT0xehljg+j+Z/Q2Fz+Y4a0=
 github.com/matrixorigin/simdcsv v0.0.0-20221007051957-5df242a514a8 h1:yr5ZAK1xhBs9i6YUNNjCXrMp2i2pA3gLH0ubNKOyxQU=
 github.com/matrixorigin/simdcsv v0.0.0-20221007051957-5df242a514a8/go.mod h1:79jCHry8mER+PwqF5EMG3ZE4b0lflH1IE/zF3Zs6lR8=
 github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
diff --git a/pkg/logservice/config.go b/pkg/logservice/config.go
index 93c688963..534a3823f 100644
--- a/pkg/logservice/config.go
+++ b/pkg/logservice/config.go
@@ -30,6 +30,7 @@ import (
 
 const (
 	defaultDataDir           = "mo-data/logservice"
+	defaultSnapshotExportDir = "exported-snapshot"
 	defaultServiceAddress    = "0.0.0.0:32000"
 	defaultRaftAddress       = "0.0.0.0:32001"
 	defaultGossipAddress     = "0.0.0.0:32002"
@@ -38,6 +39,8 @@ const (
 	defaultGossipProbeInterval = 5 * time.Second
 	defaultHeartbeatInterval   = time.Second
 	defaultLogDBBufferSize     = 768 * 1024
+	defaultTruncateInterval    = 10 * time.Second
+	defaultMaxExportedSnapshot = 20
 )
 
 // Config defines the Configurations supported by the Log Service.
@@ -56,6 +59,13 @@ type Config struct {
 	// DataDir is the name of the directory for storing all log service data. It
 	// should a locally mounted partition with good write and fsync performance.
 	DataDir string `toml:"data-dir"`
+	// SnapshotExportDir is the directory where the dragonboat snapshots are
+	// exported.
+	SnapshotExportDir string `toml:"snapshot-export-dir"`
+	// MaxExportedSnapshot is the max count of exported snapshots. If there are
+	// already MaxExportedSnapshot exported snapshots, no exported snapshot will
+	// be generated.
+	MaxExportedSnapshot int `toml:"max-exported-snapshot"`
 	// ServiceAddress is log service's service address that can be reached by
 	// other nodes such as DN nodes.
 	ServiceAddress string `toml:"logservice-address"`
@@ -90,6 +100,9 @@ type Config struct {
 	// HAKeeperCheckInterval is the interval of how often HAKeeper should run
 	// cluster health checks.
 	HAKeeperCheckInterval toml.Duration `toml:"hakeeper-check-interval"`
+	// TruncateInterval is the interval of how often log service should
+	// process truncate.
+	TruncateInterval toml.Duration `toml:"truncate-interval"`
 
 	// BootstrapConfig is the configuration specified for the bootstrapping
 	// procedure. It only needs to be specified for Log Stores selected to host
@@ -253,6 +266,9 @@ func (c *Config) Validate() error {
 	if c.GossipProbeInterval.Duration == 0 {
 		return moerr.NewBadConfig("GossipProbeInterval not set")
 	}
+	if c.TruncateInterval.Duration == 0 {
+		return moerr.NewBadConfig("TruncateInterval not set")
+	}
 	// validate BootstrapConfig
 	if c.BootstrapConfig.BootstrapCluster {
 		if c.BootstrapConfig.NumOfLogShards == 0 {
@@ -292,6 +308,12 @@ func (c *Config) Fill() {
 	if len(c.DataDir) == 0 {
 		c.DataDir = defaultDataDir
 	}
+	if len(c.SnapshotExportDir) == 0 {
+		c.SnapshotExportDir = defaultSnapshotExportDir
+	}
+	if c.MaxExportedSnapshot == 0 {
+		c.MaxExportedSnapshot = defaultMaxExportedSnapshot
+	}
 	if len(c.ServiceAddress) == 0 {
 		c.ServiceAddress = defaultServiceAddress
 		c.ServiceListenAddress = defaultServiceAddress
@@ -337,6 +359,9 @@ func (c *Config) Fill() {
 	if c.GossipProbeInterval.Duration == 0 {
 		c.GossipProbeInterval.Duration = defaultGossipProbeInterval
 	}
+	if c.TruncateInterval.Duration == 0 {
+		c.TruncateInterval.Duration = defaultTruncateInterval
+	}
 }
 
 // HAKeeperClientConfig is the config for HAKeeper clients.
diff --git a/pkg/logservice/snapshot.go b/pkg/logservice/snapshot.go
new file mode 100644
index 000000000..b20f0edf8
--- /dev/null
+++ b/pkg/logservice/snapshot.go
@@ -0,0 +1,357 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"fmt"
+	"path/filepath"
+	"sort"
+	"sync"
+
+	"github.com/lni/vfs"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+)
+
+const (
+	defaultExportedDirMode = 0755
+	snapshotPathPattern    = "snapshot-%016X"
+)
+
+// ISnapshotManager is an interface that managers snapshots.
+type ISnapshotManager interface {
+	// Init initialize snapshots by loading exported snapshots.
+	Init(shardID uint64, replicaID uint64) error
+	// Count returns the number of snapshots in the manager.
+	Count(shardID uint64, replicaID uint64) int
+	// Add adds a new snapshot for specified shard.
+	Add(shardID uint64, replicaID uint64, index uint64) error
+	// Remove removes snapshots whose index is LE than index.
+	Remove(shardID uint64, replicaID uint64, index uint64) error
+	// EvalImportSnapshot returns the source directory and index of
+	// the biggest snapshot of the shard.
+	EvalImportSnapshot(shardID uint64, replicaID uint64, index uint64) (string, uint64)
+}
+
+// ISnapshotItem is an interface that represents a snapshot item.
+type ISnapshotItem interface {
+	// Exists returns if the snapshot item exists.
+	Exists() bool
+	// Remove removes the snapshot item.
+	Remove() error
+	// Valid check the legality of the snapshot item. Only the names of
+	// the files in it are checked.
+	Valid() (bool, error)
+}
+
+// snapshotIndex is type indicates the index of snapshot.
+type snapshotIndex uint64
+
+// nodeID contains shardID and replicaID.
+type nodeID struct {
+	shardID   uint64
+	replicaID uint64
+}
+
+// snapshotItem represents a snapshot item, with index and directory in it.
+type snapshotItem struct {
+	fs    vfs.FS
+	index snapshotIndex
+	dir   string
+}
+
+var snapshotItemPool = sync.Pool{
+	New: func() interface{} {
+		return new(snapshotItem)
+	},
+}
+
+func getSnapshotItem(si snapshotItem) *snapshotItem {
+	item := snapshotItemPool.Get().(*snapshotItem)
+	*item = si
+	return item
+}
+
+func putSnapshotItem(item *snapshotItem) {
+	*item = snapshotItem{}
+	snapshotItemPool.Put(item)
+}
+
+// Exists implements the ISnapshotItem interface.
+func (si *snapshotItem) Exists() bool {
+	_, err := si.fs.Stat(si.dir)
+	return err == nil
+}
+
+// Remove implements the ISnapshotItem interface.
+func (si *snapshotItem) Remove() error {
+	return si.fs.RemoveAll(si.dir)
+}
+
+// Valid implements the ISnapshotItem interface.
+func (si *snapshotItem) Valid() (bool, error) {
+	names, err := si.fs.List(si.dir)
+	if err != nil {
+		return false, err
+	}
+	if len(names) != 2 {
+		return false, moerr.NewInternalError("file number is not correct: %d", len(names))
+	}
+	sort.Strings(names)
+	var index uint64
+	_, err = fmt.Sscanf(names[0], snapshotPathPattern+".gbsnap", &index)
+	if err != nil {
+		return false, err
+	}
+	if snapshotIndex(index) != si.index {
+		return false, moerr.NewInternalError("index of dir %d and file %d are different",
+			si.index, index)
+	}
+	if names[1] != "snapshot.metadata" {
+		return false, moerr.NewInternalError("no snapshot.metadata file")
+	}
+	return true, nil
+}
+
+// snapshotRecord contains snapshot items for the specified shard.
+type snapshotRecord struct {
+	fs     vfs.FS
+	nodeID nodeID
+	// items are sorted by .index
+	items []*snapshotItem
+}
+
+func newNodeSnapshot(fs vfs.FS, nodeID nodeID) *snapshotRecord {
+	return &snapshotRecord{
+		fs:     fs,
+		nodeID: nodeID,
+		items:  make([]*snapshotItem, 0),
+	}
+}
+
+func (ss *snapshotRecord) first() *snapshotItem {
+	if len(ss.items) == 0 {
+		return nil
+	}
+	return ss.items[0]
+}
+
+func (ss *snapshotRecord) last() *snapshotItem {
+	l := len(ss.items)
+	if l == 0 {
+		return nil
+	}
+	return ss.items[l-1]
+}
+
+func (ss *snapshotRecord) add(index snapshotIndex, dir string) error {
+	last := ss.last()
+	if last != nil && index < last.index {
+		return moerr.NewInternalError("snapshot with smaller index %d than current biggest one %d",
+			index, last.index)
+	}
+	si := getSnapshotItem(snapshotItem{fs: ss.fs, index: index, dir: dir})
+	if !si.Exists() {
+		return moerr.NewInternalError("snapshot file does not exist for shard-replica %d-%d, index %d, dir %s",
+			ss.nodeID.shardID, ss.nodeID.replicaID, index, dir)
+	}
+	v, err := si.Valid()
+	if err != nil {
+		return err
+	}
+	if v {
+		ss.items = append(ss.items, si)
+	}
+	return nil
+}
+
+// removeFirst removes the first snapshot from snapshot item list.
+func (ss *snapshotRecord) removeFirst() error {
+	if first := ss.first(); first != nil {
+		if err := first.Remove(); err != nil {
+			return err
+		}
+	}
+	putSnapshotItem(ss.items[0])
+	ss.items = ss.items[1:]
+	return nil
+}
+
+// remove the snapshots whose index is LE than the index.
+func (ss *snapshotRecord) remove(index snapshotIndex) error {
+	items := make([]*snapshotItem, len(ss.items))
+	copy(items, ss.items)
+	for _, si := range items {
+		if si.index <= index {
+			if err := ss.removeFirst(); err != nil {
+				return err
+			}
+		} else {
+			break
+		}
+	}
+	return nil
+}
+
+// snapshotManager manages the exported snapshots created by dragonboat.
+// In dragonboat, snapshot are taken with method SyncRequestSnapshot,
+// which accepts a snapshot option. By default, a new snapshot will be
+// taken at the applied index and log entries are removed at the LSN
+// parameter which is in snapshot option. LSN must be less than the applied
+// index. This works in normal state machine.
+// Unfortunately, there are two separate state machines in MO. One is
+// in logservice and the other is in DN. The snapshot cannot be taken
+// like that in this case, because when a replica starts, it applies the
+// latest snapshot, whose index is greater than truncate LSN. As a result,
+// the DN cannot read the log entries between truncate LSN and snapshot
+// index to replay.
+// The solution is, set Exported to true in snapshot option. This prevents
+// taking an active snapshot, and just export the snapshot files to an external
+// directory. When truncate LSN is greater than any snapshot index in the
+// external directory, the snapshot would be imported to system. And then,
+// the log entries can be removed safely.
+type snapshotManager struct {
+	cfg       *Config
+	snapshots map[nodeID]*snapshotRecord // shardID => *snapshotRecord
+}
+
+// newSnapshotManager makes a new snapshot.
+func newSnapshotManager(cfg *Config) *snapshotManager {
+	return &snapshotManager{
+		cfg:       cfg,
+		snapshots: make(map[nodeID]*snapshotRecord),
+	}
+}
+
+func (sm *snapshotManager) exportPath(shardID uint64, replicaID uint64) string {
+	parts := make([]string, 3)
+	dir := sm.cfg.SnapshotExportDir
+	shardPart := fmt.Sprintf("shard-%d", shardID)
+	replicaPart := fmt.Sprintf("replica-%d", replicaID)
+	parts = append(parts, dir, shardPart, replicaPart)
+	return filepath.Join(parts...)
+}
+
+func (sm *snapshotManager) snapshotPath(nodeID nodeID, index snapshotIndex) string {
+	parts := make([]string, 2)
+	snapshotPart := fmt.Sprintf(snapshotPathPattern, index)
+	parts = append(parts, sm.exportPath(nodeID.shardID, nodeID.replicaID), snapshotPart)
+	return filepath.Join(parts...)
+}
+
+func (sm *snapshotManager) prepareDir(path string) error {
+	s, err := sm.cfg.FS.Stat(path)
+	if err != nil {
+		if e := sm.cfg.FS.MkdirAll(path, defaultExportedDirMode); e != nil {
+			return e
+		}
+		return nil
+	}
+	if !s.IsDir() {
+		return moerr.NewInternalError("%s is not a dir", path)
+	}
+	return nil
+}
+
+func (sm *snapshotManager) parse(dir string) (int, error) {
+	var index int
+	_, err := fmt.Sscanf(dir, snapshotPathPattern, &index)
+	if err != nil {
+		return 0, err
+	}
+	return index, nil
+}
+
+// Init implements the ISnapshotManager interface.
+func (sm *snapshotManager) Init(shardID uint64, replicaID uint64) error {
+	path := sm.exportPath(shardID, replicaID)
+	if err := sm.prepareDir(path); err != nil {
+		return err
+	}
+	names, err := sm.cfg.FS.List(path)
+	if err != nil {
+		return err
+	}
+	indexes := make([]int, len(names))
+	for _, name := range names {
+		index, err := sm.parse(name)
+		if err != nil {
+			continue
+		}
+		indexes = append(indexes, index)
+	}
+	// The snapshots in the manager must be sorted.
+	sort.Ints(indexes)
+	for _, idx := range indexes {
+		if idx > 0 {
+			_ = sm.Add(shardID, replicaID, uint64(idx))
+		}
+	}
+	return nil
+}
+
+// Count implements the ISnapshotManager interface.
+func (sm *snapshotManager) Count(shardID uint64, replicaID uint64) int {
+	nid := nodeID{shardID: shardID, replicaID: replicaID}
+	if s, ok := sm.snapshots[nid]; ok {
+		return len(s.items)
+	}
+	return 0
+}
+
+// Add implements the ISnapshotManager interface.
+func (sm *snapshotManager) Add(shardID uint64, replicaID uint64, index uint64) error {
+	si := snapshotIndex(index)
+	nid := nodeID{shardID: shardID, replicaID: replicaID}
+	dir := sm.snapshotPath(nid, si)
+	_, ok := sm.snapshots[nid]
+	if !ok {
+		sm.snapshots[nid] = newNodeSnapshot(sm.cfg.FS, nid)
+	}
+	return sm.snapshots[nid].add(si, dir)
+}
+
+// Remove implements the ISnapshotManager interface.
+func (sm *snapshotManager) Remove(shardID uint64, replicaID uint64, index uint64) error {
+	si := snapshotIndex(index)
+	nid := nodeID{shardID: shardID, replicaID: replicaID}
+	if ss, ok := sm.snapshots[nid]; ok {
+		return ss.remove(si)
+	}
+	return nil
+}
+
+// EvalImportSnapshot implements the ISnapshotManager interface.
+func (sm *snapshotManager) EvalImportSnapshot(shardID uint64, replicaID uint64, index uint64) (string, uint64) {
+	nid := nodeID{shardID: shardID, replicaID: replicaID}
+	ss, ok := sm.snapshots[nid]
+	if !ok {
+		return "", 0
+	}
+
+	var dir string
+	var si snapshotIndex
+	for _, item := range ss.items {
+		if item == nil {
+			return "", 0
+		}
+		// Find the bigger one, break and return the smaller one.
+		if uint64(item.index) > index {
+			break
+		}
+		dir = item.dir
+		si = item.index
+	}
+	return dir, uint64(si)
+}
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index ff62dfbce..1093ae564 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -132,10 +132,10 @@ type store struct {
 
 	mu struct {
 		sync.Mutex
-		truncateCh      chan struct{}
-		pendingTruncate map[uint64]struct{}
-		metadata        metadata.LogStore
+		metadata metadata.LogStore
 	}
+	shardSnapshotInfo shardSnapshotInfo
+	snapshotMgr       *snapshotManager
 }
 
 func newLogStore(cfg Config,
@@ -161,9 +161,10 @@ func newLogStore(cfg Config,
 		stopper:       stopper.NewStopper("log-store"),
 		tickerStopper: stopper.NewStopper("hakeeper-ticker"),
 		logger:        logger,
+
+		shardSnapshotInfo: newShardSnapshotInfo(),
+		snapshotMgr:       newSnapshotManager(&cfg),
 	}
-	ls.mu.truncateCh = make(chan struct{})
-	ls.mu.pendingTruncate = make(map[uint64]struct{})
 	ls.mu.metadata = metadata.LogStore{UUID: cfg.UUID}
 	if err := ls.stopper.RunNamedTask("truncation-worker", func(ctx context.Context) {
 		logger.Info("logservice truncation worker started")
@@ -233,6 +234,9 @@ func (l *store) startReplica(shardID uint64, replicaID uint64,
 		return moerr.NewInvalidInput("shardID %d does not match DefaultHAKeeperShardID %d", shardID, hakeeper.DefaultHAKeeperShardID)
 	}
 	cfg := getRaftConfig(shardID, replicaID)
+	if err := l.snapshotMgr.Init(shardID, replicaID); err != nil {
+		panic(err)
+	}
 	if err := l.nh.StartReplica(initialReplicas, join, newStateMachine, cfg); err != nil {
 		return err
 	}
@@ -364,10 +368,6 @@ func (l *store) truncateLog(ctx context.Context,
 		l.logger.Error(fmt.Sprintf("shardID %d already truncated to index %d", shardID, result.Value))
 		return moerr.NewInvalidTruncateLsn(shardID, result.Value)
 	}
-	l.mu.Lock()
-	l.mu.pendingTruncate[shardID] = struct{}{}
-	l.mu.Unlock()
-	l.mu.truncateCh <- struct{}{}
 	return nil
 }
 
@@ -676,28 +676,6 @@ func (l *store) ticker(ctx context.Context) {
 	}
 }
 
-func (l *store) truncationWorker(ctx context.Context) {
-	defer func() {
-		l.logger.Info("truncation worker stopped")
-	}()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-l.mu.truncateCh:
-			if err := l.processTruncateLog(ctx); err != nil {
-				l.logger.Error("truncate failed", zap.Error(err))
-			}
-			select {
-			case <-ctx.Done():
-				return
-			default:
-			}
-		}
-	}
-}
-
 func (l *store) isLeaderHAKeeper() (bool, uint64, error) {
 	leaderID, term, ok, err := l.nh.GetLeaderID(hakeeper.DefaultHAKeeperShardID)
 	if err != nil {
@@ -727,49 +705,6 @@ func (l *store) hakeeperTick() {
 	}
 }
 
-// TODO: add tests for this
-func (l *store) processTruncateLog(ctx context.Context) error {
-	l.mu.Lock()
-	pendings := l.mu.pendingTruncate
-	l.mu.pendingTruncate = make(map[uint64]struct{})
-	l.mu.Unlock()
-
-	for shardID := range pendings {
-		select {
-		case <-ctx.Done():
-			return nil
-		default:
-		}
-
-		if err := func() error {
-			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-			defer cancel()
-			lsn, err := l.getTruncatedLsn(ctx, shardID)
-			if err != nil {
-				l.logger.Error("GetTruncatedIndex failed", zap.Error(err))
-				// FIXME: check error type, see whether it is a tmp one
-				return err
-			}
-			// the first 4 entries for a 3-replica raft group are tiny anyway
-			if lsn > 1 {
-				opts := dragonboat.SnapshotOption{
-					OverrideCompactionOverhead: true,
-					CompactionIndex:            lsn - 1,
-				}
-				if _, err := l.nh.SyncRequestSnapshot(ctx, shardID, opts); err != nil {
-					l.logger.Error("SyncRequestSnapshot failed", zap.Error(err))
-					// FIXME: check error type, see whether it is a tmp one
-					return err
-				}
-			}
-			return nil
-		}(); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func (l *store) getHeartbeatMessage() pb.LogStoreHeartbeat {
 	m := pb.LogStoreHeartbeat{
 		UUID:           l.id(),
diff --git a/pkg/logservice/store_metadata.go b/pkg/logservice/store_metadata.go
index 420bafbf6..991853f88 100644
--- a/pkg/logservice/store_metadata.go
+++ b/pkg/logservice/store_metadata.go
@@ -22,12 +22,10 @@ import (
 	"path/filepath"
 	"runtime"
 
-	"go.uber.org/zap"
-
 	"github.com/cockroachdb/errors/oserror"
 	"github.com/lni/vfs"
-
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
+	"go.uber.org/zap"
 )
 
 const (
@@ -299,3 +297,14 @@ func (l *store) removeMetadata(shardID uint64, replicaID uint64) {
 	l.mu.metadata.Shards = shards
 	l.mustSaveMetadata()
 }
+
+func (l *store) getReplicaID(shardID uint64) int64 {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	for _, rec := range l.mu.metadata.Shards {
+		if rec.ShardID == shardID {
+			return int64(rec.ReplicaID)
+		}
+	}
+	return -1
+}
diff --git a/pkg/logservice/truncation.go b/pkg/logservice/truncation.go
new file mode 100644
index 000000000..8687a19d0
--- /dev/null
+++ b/pkg/logservice/truncation.go
@@ -0,0 +1,238 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"context"
+	"time"
+
+	"github.com/lni/dragonboat/v4"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"go.uber.org/zap"
+)
+
+type snapshotInfo struct {
+	// truncatedLsn is LSN that has been take in snapshot manager.
+	// Log entries before it have been removed.
+	truncatedLsn uint64
+	// snapshotIndex is the latest snapshot index.
+	snapshotIndex uint64
+}
+
+type shardSnapshotInfo map[uint64]*snapshotInfo // shardID => snapshotInfo
+
+func newShardSnapshotInfo() shardSnapshotInfo {
+	return make(shardSnapshotInfo)
+}
+
+func (s shardSnapshotInfo) init(shardID uint64) {
+	s[shardID] = &snapshotInfo{
+		truncatedLsn:  0,
+		snapshotIndex: 0,
+	}
+}
+
+func (s shardSnapshotInfo) forwardTruncate(shardID uint64, lsn uint64) {
+	if _, ok := s[shardID]; !ok {
+		s.init(shardID)
+	}
+	if lsn > s[shardID].truncatedLsn {
+		s[shardID].truncatedLsn = lsn
+	}
+}
+
+func (s shardSnapshotInfo) forwardSnapshot(shardID uint64, index uint64) {
+	if _, ok := s[shardID]; !ok {
+		s.init(shardID)
+	}
+	if index > s[shardID].snapshotIndex {
+		s[shardID].snapshotIndex = index
+	}
+}
+
+func (s shardSnapshotInfo) getTruncatedLsn(shardID uint64) uint64 {
+	if _, ok := s[shardID]; !ok {
+		s.init(shardID)
+		return 0
+	}
+	return s[shardID].truncatedLsn
+}
+
+func (s shardSnapshotInfo) getSnapshotIndex(shardID uint64) uint64 {
+	if _, ok := s[shardID]; !ok {
+		s.init(shardID)
+		return 0
+	}
+	return s[shardID].snapshotIndex
+}
+
+func (l *store) truncationWorker(ctx context.Context) {
+	defer func() {
+		l.logger.Info("truncation worker stopped")
+	}()
+
+	if l.cfg.TruncateInterval.Duration == 0 {
+		panic("TruncateInterval is 0")
+	}
+	ticker := time.NewTicker(l.cfg.TruncateInterval.Duration)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := l.processTruncateLog(ctx); err != nil {
+				l.logger.Error("truncate failed", zap.Error(err))
+			}
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+		}
+	}
+}
+
+// processTruncateLog process log truncation for all shards excpet
+// hakeeper shard.
+func (l *store) processTruncateLog(ctx context.Context) error {
+	for _, shard := range l.mu.metadata.Shards {
+		if shard.ShardID == hakeeper.DefaultHAKeeperShardID {
+			continue
+		}
+		if err := l.processShardTruncateLog(ctx, shard.ShardID); err != nil {
+			l.logger.Error("process truncate log failed", zap.Error(err))
+		}
+	}
+	return nil
+}
+
+// exportSnapshot just export the snapshot but do NOT truncate logs.
+func (l *store) exportSnapshot(ctx context.Context, shardID uint64, replicaID uint64) error {
+	opts := dragonboat.SnapshotOption{
+		// set to true means just export snapshot, do NOT truncate logs.
+		Exported:   true,
+		ExportPath: l.snapshotMgr.exportPath(shardID, replicaID),
+	}
+	// Just take a snapshot and export it. This snapshot is invisible to system.
+	idx, err := l.nh.SyncRequestSnapshot(ctx, shardID, opts)
+	if err != nil {
+		l.logger.Error("request export snapshot failed", zap.Error(err))
+		return err
+	}
+	// Add the exported snapshot to snapshot manager.
+	if err := l.snapshotMgr.Add(shardID, replicaID, idx); err != nil {
+		l.logger.Error("add exported snapshot failed", zap.Error(err))
+		return err
+	}
+	// forward the snapshot index.
+	l.shardSnapshotInfo.forwardSnapshot(shardID, idx)
+	return nil
+}
+
+func (l *store) importSnapshot(
+	ctx context.Context, shardID uint64, replicaID uint64, lsn uint64, dir string,
+) error {
+	// Import a snapshot to override the snapshot in system.
+	if err := l.nh.SyncRequestImportSnapshot(ctx, shardID, replicaID, dir); err != nil {
+		l.logger.Error("import snapshot failed", zap.Error(err))
+		return err
+	}
+	// Then remove the exported snapshot in manager.
+	if err := l.snapshotMgr.Remove(shardID, replicaID, lsn); err != nil {
+		l.logger.Error("remove exported snapshots failed")
+		return err
+	}
+	// forward the truncate lsn.
+	l.shardSnapshotInfo.forwardTruncate(shardID, lsn)
+	return nil
+}
+
+// shouldProcess checks whether we should do the truncation for the shard.
+// The param lsn comes from statemachine, that client sets.
+// TODO(liubo): add more policies for truncation.
+func (l *store) shouldProcess(shardID uint64, lsn uint64) bool {
+	// the first 4 entries for a 3-replica raft group are tiny anyway
+	return lsn > 1
+}
+
+// shouldDoImport checks whether compact logs and import snapshot.
+// The param lsn is the biggest one from exported snapshots that are
+// managed by snapshotManager.
+// This check avoid doing the truncation at the same lsn.
+func (l *store) shouldDoImport(shardID uint64, lsn uint64, dir string) bool {
+	if len(dir) == 0 {
+		return false
+	}
+	return lsn > l.shardSnapshotInfo.getTruncatedLsn(shardID)
+}
+
+// shouldDoExport checks whether current applied index is greater than
+// snapshot index, which is last applied index. Means that, if applied
+// index does not advance, do not do export.
+func (l *store) shouldDoExport(ctx context.Context, shardID uint64, replicaID uint64) bool {
+	if l.snapshotMgr.Count(shardID, replicaID) >= l.cfg.MaxExportedSnapshot {
+		return false
+	}
+	v, err := l.read(ctx, shardID, indexQuery{})
+	if err != nil {
+		return false
+	}
+	return v.(uint64) > l.shardSnapshotInfo.getSnapshotIndex(shardID)
+}
+
+func (l *store) processShardTruncateLog(ctx context.Context, shardID uint64) error {
+	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+
+	lsnInSM, err := l.getTruncatedLsn(ctx, shardID)
+	if err != nil {
+		l.logger.Error("get truncated lsn in state machine failed",
+			zap.Uint64("shard ID", shardID), zap.Error(err))
+		return err
+	}
+
+	if !l.shouldProcess(shardID, lsnInSM) {
+		return nil
+	}
+
+	replicaID := uint64(l.getReplicaID(shardID))
+	dir, lsn := l.snapshotMgr.EvalImportSnapshot(shardID, replicaID, lsnInSM)
+	if l.shouldDoImport(shardID, lsn, dir) {
+		if err := l.importSnapshot(ctx, shardID, replicaID, lsn, dir); err != nil {
+			l.logger.Error("do truncate log failed",
+				zap.Uint64("shard ID", shardID),
+				zap.Uint64("replica ID", replicaID),
+				zap.Uint64("lsn", lsn),
+				zap.String("dir", dir),
+				zap.Error(err))
+			return err
+		}
+		return nil
+	}
+
+	if l.shouldDoExport(ctx, shardID, replicaID) {
+		if err := l.exportSnapshot(ctx, shardID, replicaID); err != nil {
+			l.logger.Error("export snapshot failed",
+				zap.Uint64("shard ID", shardID),
+				zap.Uint64("replica ID", replicaID),
+				zap.Uint64("lsn", lsn),
+				zap.Error(err))
+			return err
+		}
+	}
+	return nil
+}


Test patch:
diff --git a/pkg/logservice/config_test.go b/pkg/logservice/config_test.go
index 112c6700e..d94bfc42f 100644
--- a/pkg/logservice/config_test.go
+++ b/pkg/logservice/config_test.go
@@ -175,6 +175,8 @@ func TestFillConfig(t *testing.T) {
 
 	assert.Equal(t, uint64(0), c.DeploymentID)
 	assert.Equal(t, defaultDataDir, c.DataDir)
+	assert.Equal(t, defaultSnapshotExportDir, c.SnapshotExportDir)
+	assert.Equal(t, defaultMaxExportedSnapshot, c.MaxExportedSnapshot)
 	assert.Equal(t, defaultServiceAddress, c.ServiceAddress)
 	assert.Equal(t, defaultServiceAddress, c.ServiceListenAddress)
 	assert.Equal(t, defaultRaftAddress, c.RaftAddress)
diff --git a/pkg/logservice/snapshot_test.go b/pkg/logservice/snapshot_test.go
new file mode 100644
index 000000000..44185ecfe
--- /dev/null
+++ b/pkg/logservice/snapshot_test.go
@@ -0,0 +1,261 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/lni/vfs"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSnapshotItemValid(t *testing.T) {
+	si := &snapshotItem{
+		fs:    vfs.NewStrictMem(),
+		index: 100,
+		dir:   "/tmp/exported",
+	}
+	err := si.fs.MkdirAll(si.dir, defaultExportedDirMode)
+	assert.NoError(t, err)
+
+	_, err = si.fs.Create(si.dir + "/snapshot.metadata")
+	assert.NoError(t, err)
+	_, err = si.fs.Create(si.dir + "/snapshot-0000000000000064.gbsnap")
+	assert.NoError(t, err)
+
+	v, err := si.Valid()
+	assert.Equal(t, true, v)
+	assert.NoError(t, err)
+}
+
+func TestSnapshotItemInValid(t *testing.T) {
+	si := &snapshotItem{
+		fs:    vfs.NewStrictMem(),
+		index: 100,
+		dir:   "/tmp/exported",
+	}
+	err := si.fs.MkdirAll(si.dir, defaultExportedDirMode)
+	assert.NoError(t, err)
+
+	_, err = si.fs.Create(si.dir + "/snapshot.m")
+	assert.NoError(t, err)
+	_, err = si.fs.Create(si.dir + "/snapshot-0000000000000064.gbsnap")
+	assert.NoError(t, err)
+
+	v, err := si.Valid()
+	assert.Equal(t, false, v)
+	assert.Error(t, err)
+}
+
+func testPrepareSnapshot(
+	t *testing.T, mgr *snapshotManager, nid nodeID, index snapshotIndex,
+) {
+	metadataFile := "/snapshot.metadata"
+	err := mgr.cfg.FS.MkdirAll(mgr.snapshotPath(nid, index), defaultExportedDirMode)
+	assert.NoError(t, err)
+	_, err = mgr.cfg.FS.Create(mgr.snapshotPath(nid, index) + metadataFile)
+	assert.NoError(t, err)
+	_, err = mgr.cfg.FS.Create(
+		fmt.Sprintf("%s/snapshot-%016X.gbsnap", mgr.snapshotPath(nid, index), index))
+	assert.NoError(t, err)
+}
+
+func TestInitSnapshotMgr(t *testing.T) {
+	var (
+		shardID   uint64 = 1
+		replicaID uint64 = 1
+	)
+	cfg := &Config{
+		FS:                vfs.NewStrictMem(),
+		SnapshotExportDir: "/tmp/exported",
+	}
+	mgr := newSnapshotManager(cfg)
+	err := mgr.cfg.FS.MkdirAll(mgr.exportPath(shardID, replicaID), defaultExportedDirMode)
+	assert.NoError(t, err)
+
+	nid := nodeID{shardID: shardID, replicaID: replicaID}
+	indexes := []snapshotIndex{
+		snapshotIndex(310),
+		snapshotIndex(1324),
+		snapshotIndex(1211),
+		snapshotIndex(1203),
+	}
+	for _, index := range indexes {
+		testPrepareSnapshot(t, mgr, nid, index)
+	}
+	// add a bad dir
+	err = mgr.cfg.FS.MkdirAll(mgr.exportPath(shardID, replicaID)+"/bad-dir", defaultExportedDirMode)
+	assert.NoError(t, err)
+
+	err = mgr.Init(shardID, replicaID)
+	assert.NoError(t, err)
+	assert.Equal(t, 4, mgr.Count(shardID, replicaID))
+	assert.Equal(t, mgr.snapshots[nid].items[0].index, snapshotIndex(310))
+	assert.Equal(t, mgr.snapshots[nid].items[1].index, snapshotIndex(1203))
+	assert.Equal(t, mgr.snapshots[nid].items[2].index, snapshotIndex(1211))
+	assert.Equal(t, mgr.snapshots[nid].items[3].index, snapshotIndex(1324))
+}
+
+func TestAddSnapshot(t *testing.T) {
+	var (
+		shardID   uint64 = 1
+		replicaID uint64 = 1
+	)
+	cfg := &Config{
+		FS:                vfs.NewStrictMem(),
+		SnapshotExportDir: "",
+	}
+	mgr := newSnapshotManager(cfg)
+	err := mgr.cfg.FS.MkdirAll(mgr.exportPath(shardID, replicaID), defaultExportedDirMode)
+	assert.NoError(t, err)
+
+	err = mgr.Init(shardID, replicaID)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, mgr.Count(shardID, replicaID))
+
+	// add one
+	nid := nodeID{shardID: shardID, replicaID: replicaID}
+	index := snapshotIndex(621)
+	testPrepareSnapshot(t, mgr, nid, index)
+	err = mgr.Add(shardID, replicaID, uint64(index))
+	assert.NoError(t, err)
+	assert.Equal(t, 1, mgr.Count(shardID, replicaID))
+
+	// add another one, but smaller index, should get an error
+	index = snapshotIndex(96)
+	testPrepareSnapshot(t, mgr, nid, index)
+	err = mgr.Add(shardID, replicaID, uint64(index))
+	assert.Error(t, err)
+	assert.Equal(t, 1, mgr.Count(shardID, replicaID))
+
+	// add another one
+	index = snapshotIndex(794)
+	testPrepareSnapshot(t, mgr, nid, index)
+	err = mgr.Add(shardID, replicaID, uint64(index))
+	assert.NoError(t, err)
+	assert.Equal(t, 2, mgr.Count(shardID, replicaID))
+
+	// add an invalid one
+	index = snapshotIndex(795)
+	metadataFile := "/snapshot.bad"
+	err = mgr.cfg.FS.MkdirAll(mgr.snapshotPath(nid, index), defaultExportedDirMode)
+	assert.NoError(t, err)
+	_, err = mgr.cfg.FS.Create(mgr.snapshotPath(nid, index) + metadataFile)
+	assert.NoError(t, err)
+	_, err = mgr.cfg.FS.Create(
+		fmt.Sprintf("%s/snapshot-%016X.gbsnap", mgr.snapshotPath(nid, index), index))
+	assert.NoError(t, err)
+	err = mgr.Add(shardID, replicaID, uint64(index))
+	assert.Error(t, err)
+	assert.Equal(t, 2, mgr.Count(shardID, replicaID))
+
+	// add an invalid one
+	index = snapshotIndex(797)
+	metadataFile = "/snapshot.metadata"
+	err = mgr.cfg.FS.MkdirAll(mgr.snapshotPath(nid, index), defaultExportedDirMode)
+	assert.NoError(t, err)
+	_, err = mgr.cfg.FS.Create(mgr.snapshotPath(nid, index) + metadataFile)
+	assert.NoError(t, err)
+	_, err = mgr.cfg.FS.Create(
+		fmt.Sprintf("%s/snapshot.gbsnap", mgr.snapshotPath(nid, index)))
+	assert.NoError(t, err)
+	err = mgr.Add(shardID, replicaID, uint64(index))
+	assert.Error(t, err)
+	assert.Equal(t, 2, mgr.Count(shardID, replicaID))
+}
+
+func getInitializedMgr(t *testing.T, shardID uint64, replicaID uint64) *snapshotManager {
+	cfg := &Config{
+		FS:                vfs.NewStrictMem(),
+		SnapshotExportDir: "",
+	}
+	mgr := newSnapshotManager(cfg)
+	err := mgr.Init(shardID, replicaID)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, mgr.Count(shardID, replicaID))
+
+	nid := nodeID{shardID: shardID, replicaID: replicaID}
+	index := snapshotIndex(100)
+	testPrepareSnapshot(t, mgr, nid, index)
+	err = mgr.Add(shardID, replicaID, uint64(index))
+	assert.NoError(t, err)
+	assert.Equal(t, 1, mgr.Count(shardID, replicaID))
+	assert.Equal(t, mgr.snapshots[nid].items[0].index, index)
+
+	index = snapshotIndex(200)
+	testPrepareSnapshot(t, mgr, nid, index)
+	err = mgr.Add(shardID, replicaID, uint64(index))
+	assert.NoError(t, err)
+	assert.Equal(t, 2, mgr.Count(shardID, replicaID))
+	assert.Equal(t, mgr.snapshots[nid].items[1].index, index)
+
+	index = snapshotIndex(300)
+	testPrepareSnapshot(t, mgr, nid, index)
+	err = mgr.Add(shardID, replicaID, uint64(index))
+	assert.NoError(t, err)
+	assert.Equal(t, 3, mgr.Count(shardID, replicaID))
+	assert.Equal(t, mgr.snapshots[nid].items[2].index, index)
+
+	return mgr
+}
+
+func TestRemoveSnapshot(t *testing.T) {
+	var (
+		shardID   uint64 = 1
+		replicaID uint64 = 1
+	)
+	mgr := getInitializedMgr(t, shardID, replicaID)
+	err := mgr.Remove(shardID, replicaID, 200)
+	nid := nodeID{shardID: shardID, replicaID: replicaID}
+	assert.NoError(t, err)
+	assert.Equal(t, 1, mgr.Count(shardID, replicaID))
+	assert.Equal(t, mgr.snapshots[nid].items[0].index, snapshotIndex(300))
+}
+
+func TestEvalImportSnapshot(t *testing.T) {
+	var (
+		shardID   uint64 = 1
+		replicaID uint64 = 1
+	)
+	mgr := getInitializedMgr(t, shardID, replicaID)
+
+	dir, index := mgr.EvalImportSnapshot(shardID, replicaID, 50)
+	assert.Equal(t, "", dir)
+	assert.Equal(t, uint64(0), index)
+
+	dir, index = mgr.EvalImportSnapshot(shardID, replicaID, 110)
+	assert.Equal(t, "shard-1/replica-1/snapshot-0000000000000064", dir)
+	assert.Equal(t, uint64(100), index)
+
+	dir, index = mgr.EvalImportSnapshot(shardID, replicaID, 210)
+	assert.Equal(t, "shard-1/replica-1/snapshot-00000000000000C8", dir)
+	assert.Equal(t, uint64(200), index)
+
+	err := mgr.Remove(shardID, replicaID, 150)
+	assert.NoError(t, err)
+
+	dir, index = mgr.EvalImportSnapshot(shardID, replicaID, 110)
+	assert.Equal(t, "", dir)
+	assert.Equal(t, uint64(0), index)
+
+	dir, index = mgr.EvalImportSnapshot(shardID, replicaID, 210)
+	assert.Equal(t, "shard-1/replica-1/snapshot-00000000000000C8", dir)
+	assert.Equal(t, uint64(200), index)
+
+	dir, index = mgr.EvalImportSnapshot(shardID, replicaID, 10000)
+	assert.Equal(t, "shard-1/replica-1/snapshot-000000000000012C", dir)
+	assert.Equal(t, uint64(300), index)
+}
diff --git a/pkg/logservice/truncation_test.go b/pkg/logservice/truncation_test.go
new file mode 100644
index 000000000..f9bd12a3c
--- /dev/null
+++ b/pkg/logservice/truncation_test.go
@@ -0,0 +1,202 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTruncationExportSnapshot(t *testing.T) {
+	fn := func(t *testing.T, s *Service) {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+		defer cancel()
+
+		dnID := uint64(100)
+		req := pb.Request{
+			Method: pb.CONNECT_RO,
+			LogRequest: pb.LogRequest{
+				ShardID: 1,
+				DNID:    dnID,
+			},
+		}
+		resp := s.handleConnect(ctx, req)
+		assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+
+		for i := 0; i < 10; i++ {
+			data := make([]byte, 8)
+			cmd := getTestAppendCmd(dnID, data)
+			req = pb.Request{
+				Method: pb.APPEND,
+				LogRequest: pb.LogRequest{
+					ShardID: 1,
+				},
+			}
+			resp = s.handleAppend(ctx, req, cmd)
+			assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+			assert.Equal(t, uint64(4+i), resp.LogResponse.Lsn) // applied index is 4+i
+		}
+
+		req = pb.Request{
+			Method: pb.TRUNCATE,
+			LogRequest: pb.LogRequest{
+				ShardID: 1,
+				Lsn:     4,
+			},
+		}
+		resp = s.handleTruncate(ctx, req)
+		assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+		assert.Equal(t, uint64(0), resp.LogResponse.Lsn)
+
+		req = pb.Request{
+			Method: pb.GET_TRUNCATE,
+			LogRequest: pb.LogRequest{
+				ShardID: 1,
+			},
+		}
+		resp = s.handleGetTruncatedIndex(ctx, req)
+		assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+		assert.Equal(t, uint64(4), resp.LogResponse.Lsn)
+
+		err := s.store.processShardTruncateLog(ctx, 1)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, s.store.snapshotMgr.Count(1, 1))
+
+		err = s.store.processShardTruncateLog(ctx, 1)
+		// truncate lsn not advanced, no error is returned.
+		assert.NoError(t, err)
+	}
+	runServiceTest(t, false, true, fn)
+}
+
+func TestTruncationImportSnapshot(t *testing.T) {
+	fn := func(t *testing.T, s *Service) {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+		defer cancel()
+
+		dnID := uint64(100)
+		req := pb.Request{
+			Method: pb.CONNECT_RO,
+			LogRequest: pb.LogRequest{
+				ShardID: 1,
+				DNID:    dnID,
+			},
+		}
+		resp := s.handleConnect(ctx, req)
+		assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+
+		for i := 0; i < 10; i++ {
+			data := make([]byte, 8)
+			cmd := getTestAppendCmd(dnID, data)
+			req = pb.Request{
+				Method: pb.APPEND,
+				LogRequest: pb.LogRequest{
+					ShardID: 1,
+				},
+			}
+			resp = s.handleAppend(ctx, req, cmd)
+			assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+			assert.Equal(t, uint64(4+i), resp.LogResponse.Lsn) // applied index is 4+i
+		}
+
+		req = pb.Request{
+			Method: pb.TRUNCATE,
+			LogRequest: pb.LogRequest{
+				ShardID: 1,
+				Lsn:     4,
+			},
+		}
+		resp = s.handleTruncate(ctx, req)
+		assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+		assert.Equal(t, uint64(0), resp.LogResponse.Lsn)
+
+		req = pb.Request{
+			Method: pb.GET_TRUNCATE,
+			LogRequest: pb.LogRequest{
+				ShardID: 1,
+			},
+		}
+		resp = s.handleGetTruncatedIndex(ctx, req)
+		assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+		assert.Equal(t, uint64(4), resp.LogResponse.Lsn)
+
+		// after this, snapshot index 14 is exported.
+		err := s.store.processShardTruncateLog(ctx, 1)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, s.store.snapshotMgr.Count(1, 1))
+
+		_, idx := s.store.snapshotMgr.EvalImportSnapshot(1, 1, 6)
+		assert.Equal(t, uint64(0), idx)
+
+		err = s.store.processShardTruncateLog(ctx, 1)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, s.store.snapshotMgr.Count(1, 1))
+
+		req = pb.Request{
+			Method: pb.TRUNCATE,
+			LogRequest: pb.LogRequest{
+				ShardID: 1,
+				Lsn:     10,
+			},
+		}
+		resp = s.handleTruncate(ctx, req)
+		assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+		assert.Equal(t, uint64(0), resp.LogResponse.Lsn)
+
+		_, idx = s.store.snapshotMgr.EvalImportSnapshot(1, 1, 10)
+		assert.Equal(t, uint64(0), idx)
+		// index already advance to 15 because of truncate op.
+		err = s.store.processShardTruncateLog(ctx, 1)
+		assert.NoError(t, err)
+		assert.Equal(t, 2, s.store.snapshotMgr.Count(1, 1))
+
+		for i := 0; i < 10; i++ {
+			data := make([]byte, 8)
+			cmd := getTestAppendCmd(dnID, data)
+			req = pb.Request{
+				Method: pb.APPEND,
+				LogRequest: pb.LogRequest{
+					ShardID: 1,
+				},
+			}
+			resp = s.handleAppend(ctx, req, cmd)
+			assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+			assert.Equal(t, uint64(16+i), resp.LogResponse.Lsn) // applied index is 16+i
+		}
+
+		req = pb.Request{
+			Method: pb.TRUNCATE,
+			LogRequest: pb.LogRequest{
+				ShardID: 1,
+				Lsn:     15,
+			},
+		}
+		resp = s.handleTruncate(ctx, req)
+		assert.Equal(t, uint32(moerr.Ok), resp.ErrorCode)
+		assert.Equal(t, uint64(0), resp.LogResponse.Lsn)
+
+		_, idx = s.store.snapshotMgr.EvalImportSnapshot(1, 1, 14)
+		assert.Equal(t, uint64(14), idx)
+		err = s.store.processShardTruncateLog(ctx, 1)
+		assert.NoError(t, err)
+		assert.Equal(t, 0, s.store.snapshotMgr.Count(1, 1))
+	}
+	runServiceTest(t, false, true, fn)
+}