Repository: matrixorigin/matrixone
Pull request #3420 (2022-06-25 08:48:16)

logservice: added type field to LogRecord, return all entries
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

Based on Xu Peng's request during the Txn code introduction meeting, LogRecords returned by the LogService will now contain all entries including those internal ones generated by Raft & LogService. A type field has been added to indicate whether the LogRecord instance is suppose to be consumed by LogService's user code.  

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go
index 18a6da867..de728fa1e 100644
--- a/pkg/logservice/client.go
+++ b/pkg/logservice/client.go
@@ -134,11 +134,11 @@ func (c *client) connectReadWrite(ctx context.Context) error {
 	if c.readOnly() {
 		panic(ErrIncompatibleClient)
 	}
-	return c.connect(ctx, pb.MethodType_CONNECT)
+	return c.connect(ctx, pb.CONNECT)
 }
 
 func (c *client) connectReadOnly(ctx context.Context) error {
-	return c.connect(ctx, pb.MethodType_CONNECT_RO)
+	return c.connect(ctx, pb.CONNECT_RO)
 }
 
 func (c *client) request(ctx context.Context,
@@ -177,7 +177,7 @@ func (c *client) connect(ctx context.Context, mt pb.MethodType) error {
 }
 
 func (c *client) append(ctx context.Context, rec pb.LogRecord) (Lsn, error) {
-	resp, _, err := c.request(ctx, pb.MethodType_APPEND, rec.Data, 0, 0)
+	resp, _, err := c.request(ctx, pb.APPEND, rec.Data, 0, 0)
 	if err != nil {
 		return 0, err
 	}
@@ -186,7 +186,7 @@ func (c *client) append(ctx context.Context, rec pb.LogRecord) (Lsn, error) {
 
 func (c *client) read(ctx context.Context,
 	firstIndex Lsn, maxSize uint64) ([]pb.LogRecord, Lsn, error) {
-	resp, recs, err := c.request(ctx, pb.MethodType_READ, nil, firstIndex, maxSize)
+	resp, recs, err := c.request(ctx, pb.READ, nil, firstIndex, maxSize)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -194,12 +194,12 @@ func (c *client) read(ctx context.Context,
 }
 
 func (c *client) truncate(ctx context.Context, lsn Lsn) error {
-	_, _, err := c.request(ctx, pb.MethodType_TRUNCATE, nil, lsn, 0)
+	_, _, err := c.request(ctx, pb.TRUNCATE, nil, lsn, 0)
 	return err
 }
 
 func (c *client) getTruncatedIndex(ctx context.Context) (Lsn, error) {
-	resp, _, err := c.request(ctx, pb.MethodType_GET_TRUNCATE, nil, 0, 0)
+	resp, _, err := c.request(ctx, pb.GET_TRUNCATE, nil, 0, 0)
 	if err != nil {
 		return 0, err
 	}
diff --git a/pkg/logservice/errors.go b/pkg/logservice/errors.go
index 1d7b3f6d3..3b2b5dfcc 100644
--- a/pkg/logservice/errors.go
+++ b/pkg/logservice/errors.go
@@ -38,25 +38,25 @@ var errorToCodeMappings = getErrorToCodeMapping()
 
 func getErrorToCodeMapping() []errorToCode {
 	return []errorToCode{
-		{dragonboat.ErrTimeout, pb.ErrorCode_Timeout, true},
-		{dragonboat.ErrShardNotFound, pb.ErrorCode_InvalidShard, true},
+		{dragonboat.ErrTimeout, pb.Timeout, true},
+		{dragonboat.ErrShardNotFound, pb.InvalidShard, true},
 		// TODO: why ErrTimeoutTooSmall is possible
-		{dragonboat.ErrTimeoutTooSmall, pb.ErrorCode_Timeout, false},
-		{dragonboat.ErrPayloadTooBig, pb.ErrorCode_InvalidPayloadSize, true},
-		{dragonboat.ErrRejected, pb.ErrorCode_Rejected, true},
-		{dragonboat.ErrShardNotReady, pb.ErrorCode_ShardNotReady, true},
-		{dragonboat.ErrSystemBusy, pb.ErrorCode_ShardNotReady, false},
-		{dragonboat.ErrClosed, pb.ErrorCode_SystemClosed, true},
-		{dragonboat.ErrInvalidRange, pb.ErrorCode_OutOfRange, true},
+		{dragonboat.ErrTimeoutTooSmall, pb.Timeout, false},
+		{dragonboat.ErrPayloadTooBig, pb.InvalidPayloadSize, true},
+		{dragonboat.ErrRejected, pb.Rejected, true},
+		{dragonboat.ErrShardNotReady, pb.ShardNotReady, true},
+		{dragonboat.ErrSystemBusy, pb.ShardNotReady, false},
+		{dragonboat.ErrClosed, pb.SystemClosed, true},
+		{dragonboat.ErrInvalidRange, pb.OutOfRange, true},
 
-		{ErrInvalidTruncateIndex, pb.ErrorCode_IndexAlreadyTruncated, true},
-		{ErrNotLeaseHolder, pb.ErrorCode_NotLeaseHolder, true},
+		{ErrInvalidTruncateIndex, pb.IndexAlreadyTruncated, true},
+		{ErrNotLeaseHolder, pb.NotLeaseHolder, true},
 	}
 }
 
 func toErrorCode(err error) (pb.ErrorCode, string) {
 	if err == nil {
-		return pb.ErrorCode_NoError, ""
+		return pb.NoError, ""
 	}
 	for _, rec := range errorToCodeMappings {
 		if errors.Is(err, rec.err) {
@@ -65,14 +65,14 @@ func toErrorCode(err error) (pb.ErrorCode, string) {
 		}
 	}
 	plog.Errorf("unrecognized error %v, converted to %d", err,
-		pb.ErrorCode_OtherSystemError)
-	return pb.ErrorCode_OtherSystemError, err.Error()
+		pb.OtherSystemError)
+	return pb.OtherSystemError, err.Error()
 }
 
 func toError(resp pb.Response) error {
-	if resp.ErrorCode == pb.ErrorCode_NoError {
+	if resp.ErrorCode == pb.NoError {
 		return nil
-	} else if resp.ErrorCode == pb.ErrorCode_OtherSystemError {
+	} else if resp.ErrorCode == pb.OtherSystemError {
 		return errors.Wrapf(ErrUnknownError, resp.ErrorMessage)
 	}
 
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index 3a2afd62b..7a6ad2982 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -181,21 +181,21 @@ func (s *Service) serve(conn net.Conn) {
 func (s *Service) handle(req pb.Request,
 	payload []byte) (pb.Response, pb.LogRecordResponse) {
 	switch req.Method {
-	case pb.MethodType_CREATE:
+	case pb.CREATE:
 		panic("not implemented")
-	case pb.MethodType_DESTROY:
+	case pb.DESTROY:
 		panic("not implemented")
-	case pb.MethodType_APPEND:
+	case pb.APPEND:
 		return s.handleAppend(req, payload), pb.LogRecordResponse{}
-	case pb.MethodType_READ:
+	case pb.READ:
 		return s.handleRead(req)
-	case pb.MethodType_TRUNCATE:
+	case pb.TRUNCATE:
 		return s.handleTruncate(req), pb.LogRecordResponse{}
-	case pb.MethodType_GET_TRUNCATE:
+	case pb.GET_TRUNCATE:
 		return s.handleGetTruncatedIndex(req), pb.LogRecordResponse{}
-	case pb.MethodType_CONNECT:
+	case pb.CONNECT:
 		return s.handleConnect(req), pb.LogRecordResponse{}
-	case pb.MethodType_CONNECT_RO:
+	case pb.CONNECT_RO:
 		return s.handleConnectRO(req), pb.LogRecordResponse{}
 	default:
 		panic("unknown method type")
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index aac1fa806..59d2fbe54 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -378,32 +378,48 @@ func isRaftInternalEntry(e raftpb.Entry) bool {
 	return e.Type == raftpb.ConfigChangeEntry || e.Type == raftpb.MetadataEntry
 }
 
-func (l *store) filterEntries(ctx context.Context,
-	shardID uint64, entries []raftpb.Entry) ([]LogRecord, error) {
+func (l *store) markEntries(ctx context.Context,
+	shardID uint64, entries []raftpb.Entry) ([]pb.LogRecord, error) {
 	if len(entries) == 0 {
-		return []LogRecord{}, nil
+		return []pb.LogRecord{}, nil
 	}
 	leaseHolderID, err := l.getLeaseHolderID(ctx, shardID, entries)
 	if err != nil {
 		return nil, err
 	}
-	result := make([]LogRecord, 0)
+	result := make([]pb.LogRecord, 0)
 	for _, e := range entries {
 		if isRaftInternalEntry(e) {
 			// raft internal stuff
+			result = append(result, LogRecord{
+				Type:  pb.Internal,
+				Index: e.Index,
+			})
 			continue
 		}
 		cmd := l.decodeCmd(e)
 		if isSetLeaseHolderUpdate(cmd) {
 			leaseHolderID = parseLeaseHolderID(cmd)
+			result = append(result, LogRecord{
+				Type:  pb.LeaseUpdate,
+				Index: e.Index,
+			})
 			continue
 		}
 		if isUserUpdate(cmd) {
 			if parseLeaseHolderID(cmd) != leaseHolderID {
 				// lease not match, skip
+				result = append(result, LogRecord{
+					Type:  pb.LeaseUpdate,
+					Index: e.Index,
+				})
 				continue
 			}
-			result = append(result, LogRecord{Data: cmd, Index: e.Index})
+			result = append(result, LogRecord{
+				Data:  cmd,
+				Type:  pb.UserRecord,
+				Index: e.Index,
+			})
 		}
 	}
 	return result, nil
@@ -438,9 +454,9 @@ func (l *store) QueryLog(ctx context.Context, shardID uint64,
 		if v.Completed() {
 			entries, logRange := v.RaftLogs()
 			next := getNextIndex(entries, firstIndex, logRange.LastIndex)
-			results, err := l.filterEntries(ctx, shardID, entries)
+			results, err := l.markEntries(ctx, shardID, entries)
 			if err != nil {
-				plog.Errorf("filterEntries failed, %v", err)
+				plog.Errorf("markEntries failed, %v", err)
 				return nil, 0, err
 			}
 			return results, next, nil
diff --git a/pkg/pb/logservice/logservice.pb.go b/pkg/pb/logservice/logservice.pb.go
index 05a120759..a1374535e 100644
--- a/pkg/pb/logservice/logservice.pb.go
+++ b/pkg/pb/logservice/logservice.pb.go
@@ -26,14 +26,14 @@ const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
 type MethodType int32
 
 const (
-	MethodType_CREATE       MethodType = 0
-	MethodType_DESTROY      MethodType = 1
-	MethodType_APPEND       MethodType = 2
-	MethodType_READ         MethodType = 3
-	MethodType_TRUNCATE     MethodType = 4
-	MethodType_GET_TRUNCATE MethodType = 5
-	MethodType_CONNECT      MethodType = 6
-	MethodType_CONNECT_RO   MethodType = 7
+	CREATE       MethodType = 0
+	DESTROY      MethodType = 1
+	APPEND       MethodType = 2
+	READ         MethodType = 3
+	TRUNCATE     MethodType = 4
+	GET_TRUNCATE MethodType = 5
+	CONNECT      MethodType = 6
+	CONNECT_RO   MethodType = 7
 )
 
 var MethodType_name = map[int32]string{
@@ -66,23 +66,54 @@ func (MethodType) EnumDescriptor() ([]byte, []int) {
 	return fileDescriptor_fd1040c5381ab5a7, []int{0}
 }
 
+type RecordType int32
+
+const (
+	UserRecord    RecordType = 0
+	Internal      RecordType = 1
+	LeaseUpdate   RecordType = 2
+	LeaseRejected RecordType = 3
+)
+
+var RecordType_name = map[int32]string{
+	0: "UserRecord",
+	1: "Internal",
+	2: "LeaseUpdate",
+	3: "LeaseRejected",
+}
+
+var RecordType_value = map[string]int32{
+	"UserRecord":    0,
+	"Internal":      1,
+	"LeaseUpdate":   2,
+	"LeaseRejected": 3,
+}
+
+func (x RecordType) String() string {
+	return proto.EnumName(RecordType_name, int32(x))
+}
+
+func (RecordType) EnumDescriptor() ([]byte, []int) {
+	return fileDescriptor_fd1040c5381ab5a7, []int{1}
+}
+
 type ErrorCode int32
 
 const (
-	ErrorCode_NoError               ErrorCode = 0
-	ErrorCode_Timeout               ErrorCode = 1
-	ErrorCode_Canceled              ErrorCode = 2
-	ErrorCode_InvalidShard          ErrorCode = 3
-	ErrorCode_InvalidTimeout        ErrorCode = 4
-	ErrorCode_InvalidPayload        ErrorCode = 5
-	ErrorCode_InvalidPayloadSize    ErrorCode = 6
-	ErrorCode_Rejected              ErrorCode = 7
-	ErrorCode_ShardNotReady         ErrorCode = 8
-	ErrorCode_SystemClosed          ErrorCode = 9
-	ErrorCode_IndexAlreadyTruncated ErrorCode = 100
-	ErrorCode_OutOfRange            ErrorCode = 101
-	ErrorCode_NotLeaseHolder        ErrorCode = 102
-	ErrorCode_OtherSystemError      ErrorCode = 1000
+	NoError               ErrorCode = 0
+	Timeout               ErrorCode = 1
+	Canceled              ErrorCode = 2
+	InvalidShard          ErrorCode = 3
+	InvalidTimeout        ErrorCode = 4
+	InvalidPayload        ErrorCode = 5
+	InvalidPayloadSize    ErrorCode = 6
+	Rejected              ErrorCode = 7
+	ShardNotReady         ErrorCode = 8
+	SystemClosed          ErrorCode = 9
+	IndexAlreadyTruncated ErrorCode = 100
+	OutOfRange            ErrorCode = 101
+	NotLeaseHolder        ErrorCode = 102
+	OtherSystemError      ErrorCode = 1000
 )
 
 var ErrorCode_name = map[int32]string{
@@ -124,7 +155,7 @@ func (x ErrorCode) String() string {
 }
 
 func (ErrorCode) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_fd1040c5381ab5a7, []int{1}
+	return fileDescriptor_fd1040c5381ab5a7, []int{2}
 }
 
 // LogShardInfo contains information a log shard.
@@ -541,12 +572,14 @@ func (m *RSMState) GetLeaseHistory() map[uint64]uint64 {
 	return nil
 }
 
+// LogRecord is what we store into the LogService.
 type LogRecord struct {
-	Index                uint64   `protobuf:"varint,1,opt,name=Index,proto3" json:"Index,omitempty"`
-	Data                 []byte   `protobuf:"bytes,2,opt,name=Data,proto3" json:"Data,omitempty"`
-	XXX_NoUnkeyedLiteral struct{} `json:"-"`
-	XXX_unrecognized     []byte   `json:"-"`
-	XXX_sizecache        int32    `json:"-"`
+	Index                uint64     `protobuf:"varint,1,opt,name=Index,proto3" json:"Index,omitempty"`
+	Type                 RecordType `protobuf:"varint,2,opt,name=Type,proto3,enum=logservice.RecordType" json:"Type,omitempty"`
+	Data                 []byte     `protobuf:"bytes,3,opt,name=Data,proto3" json:"Data,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}   `json:"-"`
+	XXX_unrecognized     []byte     `json:"-"`
+	XXX_sizecache        int32      `json:"-"`
 }
 
 func (m *LogRecord) Reset()         { *m = LogRecord{} }
@@ -589,6 +622,13 @@ func (m *LogRecord) GetIndex() uint64 {
 	return 0
 }
 
+func (m *LogRecord) GetType() RecordType {
+	if m != nil {
+		return m.Type
+	}
+	return UserRecord
+}
+
 func (m *LogRecord) GetData() []byte {
 	if m != nil {
 		return m.Data
@@ -648,7 +688,7 @@ func (m *Request) GetMethod() MethodType {
 	if m != nil {
 		return m.Method
 	}
-	return MethodType_CREATE
+	return CREATE
 }
 
 func (m *Request) GetName() string {
@@ -757,14 +797,14 @@ func (m *Response) GetMethod() MethodType {
 	if m != nil {
 		return m.Method
 	}
-	return MethodType_CREATE
+	return CREATE
 }
 
 func (m *Response) GetErrorCode() ErrorCode {
 	if m != nil {
 		return m.ErrorCode
 	}
-	return ErrorCode_NoError
+	return NoError
 }
 
 func (m *Response) GetErrorMessage() string {
@@ -851,6 +891,7 @@ func (m *LogRecordResponse) GetRecords() []LogRecord {
 
 func init() {
 	proto.RegisterEnum("logservice.MethodType", MethodType_name, MethodType_value)
+	proto.RegisterEnum("logservice.RecordType", RecordType_name, RecordType_value)
 	proto.RegisterEnum("logservice.ErrorCode", ErrorCode_name, ErrorCode_value)
 	proto.RegisterType((*LogShardInfo)(nil), "logservice.LogShardInfo")
 	proto.RegisterMapType((map[uint64]string)(nil), "logservice.LogShardInfo.ReplicasEntry")
@@ -869,69 +910,73 @@ func init() {
 func init() { proto.RegisterFile("logservice.proto", fileDescriptor_fd1040c5381ab5a7) }
 
 var fileDescriptor_fd1040c5381ab5a7 = []byte{
-	// 990 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x56, 0xcf, 0x6b, 0x1b, 0x47,
-	0x14, 0xce, 0x4a, 0xab, 0x5f, 0xcf, 0xb2, 0x18, 0x0f, 0x75, 0xba, 0x35, 0xc1, 0x31, 0xa2, 0x04,
-	0x13, 0xa8, 0x0c, 0x36, 0x86, 0xd2, 0x16, 0x8a, 0xac, 0xdd, 0x26, 0x0e, 0xf2, 0xca, 0x8c, 0xe4,
-	0x43, 0x0b, 0x25, 0x8c, 0xb4, 0xa3, 0xd5, 0x36, 0xd2, 0x8e, 0xba, 0x3b, 0x32, 0x56, 0xaf, 0xa5,
-	0xd7, 0xfe, 0x5d, 0x39, 0xe6, 0xd2, 0x6b, 0x28, 0x3e, 0xf5, 0xd6, 0x3f, 0xa0, 0x97, 0x32, 0x33,
-	0xbb, 0xd2, 0xc8, 0xc6, 0x6d, 0x72, 0x9b, 0xf7, 0xcd, 0xfb, 0xf9, 0xcd, 0xf7, 0x96, 0x05, 0x34,
-	0xe5, 0x61, 0xca, 0x92, 0xeb, 0x68, 0xc4, 0x5a, 0xf3, 0x84, 0x0b, 0x8e, 0x61, 0x8d, 0xec, 0x7d,
-	0x11, 0x46, 0x62, 0xb2, 0x18, 0xb6, 0x46, 0x7c, 0x76, 0x14, 0xf2, 0x90, 0x1f, 0x29, 0x97, 0xe1,
-	0x62, 0xac, 0x2c, 0x65, 0xa8, 0x93, 0x0e, 0x6d, 0xfe, 0x6d, 0x41, 0xbd, 0xcb, 0xc3, 0xfe, 0x84,
-	0x26, 0xc1, 0x79, 0x3c, 0xe6, 0xd8, 0x81, 0x8a, 0x36, 0x5c, 0xc7, 0x3a, 0xb0, 0x0e, 0x6d, 0x92,
-	0x9b, 0xf8, 0x0c, 0xaa, 0x84, 0xcd, 0xa7, 0xd1, 0x88, 0xa6, 0x4e, 0xe1, 0xa0, 0x78, 0xb8, 0x75,
-	0xfc, 0xac, 0x65, 0xb4, 0x62, 0x66, 0x69, 0xe5, 0x8e, 0x5e, 0x2c, 0x92, 0x25, 0x59, 0xc5, 0xe1,
-	0x4f, 0xa0, 0xe4, 0xcd, 0xf9, 0x68, 0xe2, 0x14, 0x55, 0x6e, 0x6d, 0xe0, 0x3d, 0xa8, 0x76, 0x19,
-	0x0d, 0x58, 0x72, 0xee, 0x3a, 0xb6, 0xba, 0x58, 0xd9, 0x18, 0x83, 0x3d, 0x60, 0xc9, 0xcc, 0x29,
-	0x29, 0x5c, 0x9d, 0xf7, 0xbe, 0x86, 0xed, 0x8d, 0x02, 0x18, 0x41, 0xf1, 0x0d, 0x5b, 0x66, 0x0d,
-	0xcb, 0xa3, 0x2c, 0x74, 0x4d, 0xa7, 0x0b, 0xe6, 0x14, 0x0e, 0xac, 0xc3, 0x1a, 0xd1, 0xc6, 0x57,
-	0x85, 0x2f, 0xad, 0xe6, 0x35, 0x34, 0xba, 0x3c, 0xcc, 0xe2, 0xd5, 0xc8, 0xdf, 0x6d, 0x52, 0xa0,
-	0xd2, 0x6c, 0x1d, 0x3b, 0x0f, 0x0d, 0x77, 0x56, 0x7d, 0xfb, 0xfe, 0xe9, 0xa3, 0x77, 0xef, 0x9f,
-	0x5a, 0x64, 0x93, 0xba, 0x27, 0x50, 0xcb, 0xd3, 0xba, 0xaa, 0xae, 0x4d, 0xd6, 0x40, 0xf3, 0x0f,
-	0x0b, 0x76, 0xa4, 0xbb, 0xe0, 0x09, 0x7b, 0xc9, 0x68, 0x22, 0x86, 0x8c, 0x0a, 0x39, 0xde, 0xd5,
-	0x55, 0xc6, 0x75, 0x8d, 0xa8, 0x33, 0x3e, 0x80, 0x2d, 0x42, 0xc7, 0xa2, 0x1d, 0x04, 0x09, 0x4b,
-	0xd3, 0x6c, 0x02, 0x13, 0xc2, 0xcf, 0xa0, 0xd1, 0xd7, 0x9d, 0xe5, 0x4e, 0x45, 0xe5, 0x74, 0x07,
-	0xc5, 0x9f, 0xc3, 0xf6, 0x0b, 0x9e, 0xa6, 0xd1, 0x3c, 0x77, 0xb3, 0x95, 0xdb, 0x26, 0x88, 0xbf,
-	0x31, 0x1e, 0xb6, 0xa4, 0x1e, 0x76, 0xef, 0xce, 0xec, 0x06, 0x5b, 0x67, 0xb6, 0x9c, 0x7e, 0xfd,
-	0xa4, 0x4d, 0x0f, 0xb6, 0x5c, 0xff, 0x43, 0xf4, 0xf3, 0xdf, 0xf4, 0xfc, 0x08, 0xc8, 0xf5, 0x3f,
-	0x80, 0x9c, 0x53, 0x28, 0xab, 0x84, 0xb9, 0x06, 0x3f, 0x35, 0x5b, 0x35, 0x1a, 0xc9, 0xfa, 0xcc,
-	0x9c, 0x9b, 0xff, 0x58, 0x50, 0x25, 0xfd, 0x8b, 0xbe, 0xa0, 0x82, 0x49, 0x71, 0x9c, 0xc7, 0x01,
-	0xbb, 0xc9, 0x3a, 0xd4, 0x86, 0x24, 0xab, 0xcb, 0x68, 0xca, 0x5e, 0xf2, 0xa9, 0x96, 0xa2, 0xee,
-	0x71, 0x13, 0x94, 0xd4, 0x0f, 0x92, 0x45, 0x3c, 0xa2, 0x82, 0x05, 0x3a, 0x89, 0x96, 0xf2, 0x1d,
-	0x14, 0xbf, 0x82, 0xba, 0x0e, 0x8c, 0x52, 0xc1, 0x93, 0xa5, 0x63, 0xdf, 0xdf, 0x98, 0xbc, 0x9f,
-	0x96, 0xe9, 0xa8, 0x37, 0x66, 0x23, 0x76, 0xef, 0x5b, 0xd8, 0xb9, 0xe7, 0xf2, 0x7f, 0x9a, 0xb7,
-	0x4d, 0xcd, 0x9f, 0x42, 0x4d, 0xbd, 0xe2, 0x88, 0x27, 0xc1, 0x03, 0xd3, 0x63, 0xb0, 0x5d, 0x2a,
-	0xa8, 0x8a, 0xad, 0x13, 0x75, 0x6e, 0xfe, 0x56, 0x80, 0x0a, 0x61, 0x3f, 0x2f, 0x58, 0x2a, 0x70,
-	0x0b, 0xca, 0x17, 0x4c, 0x4c, 0x78, 0xa0, 0xc2, 0x1a, 0xc7, 0x8f, 0xcd, 0x49, 0xf4, 0xcd, 0x60,
-	0x39, 0x67, 0x24, 0xf3, 0x92, 0xf9, 0x7c, 0x3a, 0xcb, 0xf7, 0x4f, 0x9d, 0x4d, 0x6d, 0x14, 0x37,
-	0xb5, 0xb1, 0xea, 0xc9, 0x36, 0x7b, 0x72, 0xa0, 0x72, 0x41, 0x6f, 0xfa, 0xd1, 0x2f, 0x2c, 0x5b,
-	0xff, 0xdc, 0x94, 0x37, 0x83, 0x68, 0xc6, 0xf8, 0x42, 0x38, 0xe5, 0x03, 0xeb, 0xb0, 0x48, 0x72,
-	0x53, 0xaa, 0x2c, 0x57, 0x81, 0xeb, 0x54, 0xb4, 0xca, 0x56, 0x80, 0x9a, 0xd2, 0x3f, 0x77, 0x9d,
-	0xaa, 0xfe, 0x9a, 0xc8, 0xb3, 0x5c, 0xb7, 0x4b, 0xba, 0x9c, 0x72, 0x1a, 0xa8, 0x4a, 0x35, 0x75,
-	0x65, 0x42, 0xcd, 0x5f, 0x0b, 0x72, 0x43, 0xd2, 0x39, 0x8f, 0x53, 0xf6, 0xd1, 0x44, 0x9c, 0x40,
-	0xcd, 0x4b, 0x12, 0x9e, 0x74, 0x78, 0xa0, 0xd9, 0x68, 0x1c, 0xef, 0x9a, 0x21, 0xab, 0x4b, 0xb2,
-	0xf6, 0xc3, 0x4d, 0xa8, 0x2b, 0xe3, 0x82, 0xa5, 0x29, 0x0d, 0x59, 0xb6, 0xde, 0x1b, 0x98, 0xc9,
-	0xa6, 0xfd, 0x00, 0x9b, 0x25, 0x93, 0xcd, 0x27, 0x50, 0xeb, 0xd2, 0x54, 0xe8, 0x9b, 0xb2, 0x66,
-	0x66, 0x05, 0xdc, 0x65, 0xa1, 0x72, 0x9f, 0x85, 0x57, 0xea, 0xfb, 0xa5, 0x45, 0xb4, 0x62, 0xe3,
-	0x54, 0x2a, 0x44, 0x22, 0xa9, 0x63, 0x29, 0x85, 0xef, 0xde, 0xfb, 0x74, 0xc8, 0xdb, 0x6c, 0x1b,
-	0x73, 0xdf, 0xe7, 0x37, 0x00, 0x6b, 0xaa, 0x30, 0x40, 0xb9, 0x43, 0xbc, 0xf6, 0xc0, 0x43, 0x8f,
-	0xf0, 0x16, 0x54, 0x5c, 0xaf, 0x3f, 0x20, 0xbd, 0xef, 0x91, 0x25, 0x2f, 0xda, 0x97, 0x97, 0x9e,
-	0xef, 0xa2, 0x02, 0xae, 0x82, 0x4d, 0xbc, 0xb6, 0x8b, 0x8a, 0xb8, 0x0e, 0xd5, 0x01, 0xb9, 0xf2,
-	0x3b, 0x32, 0xc0, 0xc6, 0x08, 0xea, 0x2f, 0xbc, 0xc1, 0xeb, 0x15, 0x52, 0x92, 0x29, 0x3a, 0x3d,
-	0xdf, 0xf7, 0x3a, 0x03, 0x54, 0xc6, 0x0d, 0x80, 0xcc, 0x78, 0x4d, 0x7a, 0xa8, 0xf2, 0xfc, 0xf7,
-	0x82, 0xf1, 0x1e, 0xd2, 0xd5, 0xe7, 0xca, 0xd4, 0xa5, 0x33, 0x15, 0x21, 0x4b, 0x16, 0xe9, 0xd0,
-	0x78, 0xc4, 0xa6, 0x2c, 0x40, 0x05, 0x59, 0xe4, 0x3c, 0xbe, 0xa6, 0xd3, 0x28, 0x50, 0x1c, 0xa3,
-	0x22, 0xc6, 0xd0, 0xc8, 0x90, 0x3c, 0xc6, 0x36, 0xb0, 0x8c, 0x37, 0x54, 0xc2, 0x8f, 0x01, 0x6f,
-	0x62, 0x92, 0x4b, 0x54, 0x96, 0xf9, 0x09, 0xfb, 0x89, 0x8d, 0x04, 0x0b, 0x50, 0x05, 0xef, 0xc0,
-	0xb6, 0x4a, 0xec, 0x73, 0x41, 0x18, 0x0d, 0x96, 0xa8, 0x2a, 0x4b, 0xf6, 0x97, 0xa9, 0x60, 0xb3,
-	0xce, 0x94, 0xa7, 0x2c, 0x40, 0x35, 0xfc, 0x19, 0xec, 0xaa, 0xb7, 0x6a, 0x4f, 0x13, 0xe9, 0xb3,
-	0xfa, 0xe0, 0xa0, 0x40, 0x4e, 0xd9, 0x5b, 0x88, 0xde, 0x98, 0xd0, 0x38, 0x64, 0x88, 0xc9, 0x4e,
-	0x7c, 0x2e, 0x8c, 0x2f, 0x17, 0x1a, 0xe3, 0x5d, 0x40, 0x3d, 0x31, 0x61, 0x89, 0xce, 0xaa, 0x87,
-	0xfe, 0xab, 0x72, 0xd6, 0x7e, 0x7b, 0xbb, 0x6f, 0xbd, 0xbb, 0xdd, 0xb7, 0xfe, 0xbc, 0xdd, 0xb7,
-	0x7e, 0x38, 0x31, 0x7e, 0x1f, 0x66, 0x54, 0x24, 0xd1, 0x0d, 0x4f, 0xa2, 0x30, 0x8a, 0x73, 0x23,
-	0x66, 0x47, 0xf3, 0x37, 0xe1, 0xd1, 0x7c, 0x78, 0xb4, 0x7e, 0xe6, 0x61, 0x59, 0xfd, 0x4b, 0x9c,
-	0xfc, 0x1b, 0x00, 0x00, 0xff, 0xff, 0x05, 0x15, 0x41, 0x99, 0x9a, 0x08, 0x00, 0x00,
+	// 1048 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x56, 0x4f, 0x6b, 0x1b, 0x47,
+	0x14, 0xf7, 0x4a, 0xab, 0x7f, 0x4f, 0x8a, 0x3a, 0x1e, 0xea, 0x74, 0x6b, 0x82, 0x63, 0x44, 0x09,
+	0xc6, 0x50, 0x19, 0x6c, 0x02, 0xa5, 0x2d, 0x14, 0x5b, 0xbb, 0x4d, 0x14, 0xec, 0x95, 0x19, 0xc9,
+	0x87, 0x16, 0x4a, 0x18, 0x6b, 0xc7, 0xf2, 0x36, 0xd2, 0x8e, 0x3a, 0x3b, 0x32, 0x56, 0xaf, 0xa5,
+	0xd7, 0x7e, 0xae, 0x1c, 0x7d, 0xe9, 0x35, 0xb4, 0x3e, 0xf5, 0xd6, 0x0f, 0xd0, 0x4b, 0x99, 0x99,
+	0xdd, 0xd5, 0xae, 0x8d, 0xdb, 0xe4, 0x36, 0xef, 0xf7, 0xfe, 0xcc, 0x7b, 0xbf, 0xf7, 0xe6, 0x31,
+	0x80, 0xa6, 0x7c, 0x12, 0x33, 0x71, 0x15, 0x8e, 0x59, 0x77, 0x2e, 0xb8, 0xe4, 0x18, 0x56, 0xc8,
+	0xe6, 0xe7, 0x93, 0x50, 0x5e, 0x2e, 0xce, 0xbb, 0x63, 0x3e, 0xdb, 0x9b, 0xf0, 0x09, 0xdf, 0xd3,
+	0x26, 0xe7, 0x8b, 0x0b, 0x2d, 0x69, 0x41, 0x9f, 0x8c, 0x6b, 0xe7, 0x6f, 0x0b, 0x5a, 0xc7, 0x7c,
+	0x32, 0xbc, 0xa4, 0x22, 0xe8, 0x47, 0x17, 0x1c, 0x3b, 0x50, 0x33, 0x82, 0xeb, 0x58, 0xdb, 0xd6,
+	0x8e, 0x4d, 0x52, 0x11, 0x1f, 0x41, 0x9d, 0xb0, 0xf9, 0x34, 0x1c, 0xd3, 0xd8, 0x29, 0x6d, 0x97,
+	0x77, 0x9a, 0xfb, 0xcf, 0xba, 0xb9, 0x54, 0xf2, 0x51, 0xba, 0xa9, 0xa1, 0x17, 0x49, 0xb1, 0x24,
+	0x99, 0x1f, 0xfe, 0x18, 0x2a, 0xde, 0x9c, 0x8f, 0x2f, 0x9d, 0xb2, 0x8e, 0x6d, 0x04, 0xbc, 0x09,
+	0xf5, 0x63, 0x46, 0x03, 0x26, 0xfa, 0xae, 0x63, 0x6b, 0x45, 0x26, 0x63, 0x0c, 0xf6, 0x88, 0x89,
+	0x99, 0x53, 0xd1, 0xb8, 0x3e, 0x6f, 0x7e, 0x05, 0x8f, 0x0a, 0x17, 0x60, 0x04, 0xe5, 0x37, 0x6c,
+	0x99, 0x24, 0xac, 0x8e, 0xea, 0xa2, 0x2b, 0x3a, 0x5d, 0x30, 0xa7, 0xb4, 0x6d, 0xed, 0x34, 0x88,
+	0x11, 0xbe, 0x2c, 0x7d, 0x61, 0x75, 0xae, 0xa0, 0x7d, 0xcc, 0x27, 0x89, 0xbf, 0x2e, 0xf9, 0xdb,
+	0x22, 0x05, 0x3a, 0x4c, 0x73, 0xdf, 0x79, 0xa8, 0xb8, 0xa3, 0xfa, 0xdb, 0x77, 0x4f, 0xd7, 0x6e,
+	0xde, 0x3d, 0xb5, 0x48, 0x91, 0xba, 0x27, 0xd0, 0x48, 0xc3, 0xba, 0xfa, 0x5e, 0x9b, 0xac, 0x80,
+	0xce, 0xef, 0x16, 0xac, 0x2b, 0x73, 0xc9, 0x05, 0x7b, 0xc9, 0xa8, 0x90, 0xe7, 0x8c, 0x4a, 0x55,
+	0xde, 0xd9, 0x59, 0xc2, 0x75, 0x83, 0xe8, 0x33, 0xde, 0x86, 0x26, 0xa1, 0x17, 0xf2, 0x30, 0x08,
+	0x04, 0x8b, 0xe3, 0xa4, 0x82, 0x3c, 0x84, 0x9f, 0x41, 0x7b, 0x68, 0x32, 0x4b, 0x8d, 0xca, 0xda,
+	0xe8, 0x0e, 0x8a, 0x3f, 0x83, 0x47, 0x2f, 0x78, 0x1c, 0x87, 0xf3, 0xd4, 0xcc, 0xd6, 0x66, 0x45,
+	0x10, 0x7f, 0x9d, 0x6b, 0x6c, 0x45, 0x37, 0x76, 0xf3, 0x4e, 0xed, 0x39, 0xb6, 0x8e, 0x6c, 0x55,
+	0xfd, 0xaa, 0xa5, 0x1d, 0x0f, 0x9a, 0xae, 0xff, 0x3e, 0xf3, 0xf3, 0xdf, 0xf4, 0xfc, 0x00, 0xc8,
+	0xf5, 0xdf, 0x83, 0x9c, 0xe7, 0x50, 0xd5, 0x01, 0xd3, 0x19, 0xfc, 0x24, 0x9f, 0x6a, 0x2e, 0x91,
+	0x24, 0xcf, 0xc4, 0xb8, 0xf3, 0x8f, 0x05, 0x75, 0x32, 0x3c, 0x19, 0x4a, 0x2a, 0x99, 0x1a, 0x8e,
+	0x7e, 0x14, 0xb0, 0xeb, 0x24, 0x43, 0x23, 0x28, 0xb2, 0x8e, 0x19, 0x8d, 0xd9, 0x4b, 0x3e, 0x35,
+	0xa3, 0x68, 0x72, 0x2c, 0x82, 0x8a, 0xfa, 0x91, 0x58, 0x44, 0x63, 0x2a, 0x59, 0x60, 0x82, 0x98,
+	0x51, 0xbe, 0x83, 0xe2, 0x57, 0xd0, 0x32, 0x8e, 0x61, 0x2c, 0xb9, 0x58, 0x3a, 0xf6, 0xfd, 0x17,
+	0x93, 0xe6, 0xd3, 0xcd, 0x1b, 0x9a, 0x17, 0x53, 0xf0, 0xdd, 0xfc, 0x06, 0xd6, 0xef, 0x99, 0xfc,
+	0xdf, 0xcc, 0xdb, 0xf9, 0x99, 0xa7, 0xd0, 0xd0, 0x5d, 0x1c, 0x73, 0x11, 0x3c, 0x50, 0xfd, 0x2e,
+	0xd8, 0xa3, 0xe5, 0xdc, 0xf8, 0xb6, 0xf7, 0x1f, 0x17, 0xf2, 0xd4, 0x7e, 0x4a, 0x4b, 0xb4, 0x8d,
+	0xea, 0x8b, 0x4b, 0x25, 0xd5, 0x95, 0xb7, 0x88, 0x3e, 0x77, 0x7e, 0x2d, 0x41, 0x8d, 0xb0, 0x9f,
+	0x16, 0x2c, 0x96, 0xb8, 0x0b, 0xd5, 0x13, 0x26, 0x2f, 0x79, 0xa0, 0xaf, 0xb8, 0x13, 0xcd, 0x68,
+	0x74, 0xb4, 0xc4, 0x4a, 0xc5, 0xf3, 0xe9, 0x2c, 0x7d, 0xab, 0xfa, 0x9c, 0x9f, 0xa3, 0x72, 0x71,
+	0x8e, 0xb2, 0xfc, 0xed, 0x7c, 0xfe, 0x0e, 0xd4, 0x4e, 0xe8, 0xf5, 0x30, 0xfc, 0x99, 0x25, 0xab,
+	0x22, 0x15, 0x95, 0x66, 0x14, 0xce, 0x18, 0x5f, 0x48, 0xa7, 0xba, 0x6d, 0xed, 0x94, 0x49, 0x2a,
+	0xaa, 0x89, 0x4c, 0x27, 0xc6, 0x75, 0x6a, 0x66, 0x22, 0x33, 0x40, 0x57, 0xe9, 0xf7, 0x5d, 0xa7,
+	0x6e, 0x36, 0x8f, 0x3a, 0xab, 0xa7, 0x79, 0x4a, 0x97, 0x53, 0x4e, 0x03, 0x7d, 0x53, 0x43, 0xab,
+	0xf2, 0x50, 0xe7, 0x97, 0x92, 0x7a, 0x4d, 0xf1, 0x9c, 0x47, 0x31, 0xfb, 0x60, 0x22, 0x0e, 0xa0,
+	0xe1, 0x09, 0xc1, 0x45, 0x8f, 0x07, 0x69, 0x27, 0x36, 0xf2, 0x2e, 0x99, 0x92, 0xac, 0xec, 0x70,
+	0x07, 0x5a, 0x5a, 0x38, 0x61, 0x71, 0x4c, 0x27, 0x2c, 0x59, 0x05, 0x05, 0x2c, 0xcf, 0xa6, 0xfd,
+	0x00, 0x9b, 0x95, 0x3c, 0x9b, 0x4f, 0xa0, 0x71, 0x4c, 0x63, 0x69, 0x34, 0x55, 0xc3, 0x4c, 0x06,
+	0xdc, 0x65, 0xa1, 0x76, 0x9f, 0x85, 0x57, 0x7a, 0xd7, 0x99, 0xc1, 0xc9, 0xd8, 0x78, 0xae, 0x26,
+	0x44, 0x21, 0xb1, 0x63, 0xe9, 0xd7, 0xb0, 0x71, 0x6f, 0xcd, 0x28, 0x6d, 0xf2, 0x72, 0x53, 0xdb,
+	0xdd, 0x6b, 0x80, 0x15, 0x55, 0x18, 0xa0, 0xda, 0x23, 0xde, 0xe1, 0xc8, 0x43, 0x6b, 0xb8, 0x09,
+	0x35, 0xd7, 0x1b, 0x8e, 0xc8, 0xe0, 0x3b, 0x64, 0x29, 0xc5, 0xe1, 0xe9, 0xa9, 0xe7, 0xbb, 0xa8,
+	0x84, 0xeb, 0x60, 0x13, 0xef, 0xd0, 0x45, 0x65, 0xdc, 0x82, 0xfa, 0x88, 0x9c, 0xf9, 0x3d, 0xe5,
+	0x60, 0x63, 0x04, 0xad, 0x17, 0xde, 0xe8, 0x75, 0x86, 0x54, 0x54, 0x88, 0xde, 0xc0, 0xf7, 0xbd,
+	0xde, 0x08, 0x55, 0x71, 0x1b, 0x20, 0x11, 0x5e, 0x93, 0x01, 0xaa, 0xed, 0xfa, 0x00, 0xab, 0xd9,
+	0x57, 0xda, 0xb3, 0x98, 0x09, 0x83, 0xa0, 0x35, 0x15, 0xba, 0x1f, 0x49, 0x26, 0x22, 0x3a, 0x45,
+	0x16, 0xfe, 0x08, 0x9a, 0xfa, 0x8d, 0x9e, 0xcd, 0x03, 0x2a, 0x19, 0x2a, 0xe1, 0xf5, 0x64, 0x9d,
+	0x10, 0xf6, 0x23, 0x1b, 0x4b, 0x16, 0xa0, 0xf2, 0xee, 0x6f, 0xa5, 0x5c, 0x7f, 0xd5, 0xd5, 0x3e,
+	0xd7, 0xa2, 0x29, 0x25, 0x99, 0x4a, 0x64, 0xa9, 0xc8, 0x3d, 0x1a, 0x8d, 0xd9, 0x94, 0x05, 0xa8,
+	0xa4, 0x92, 0xee, 0x47, 0x57, 0x74, 0x1a, 0x06, 0xba, 0x67, 0xa8, 0x8c, 0x31, 0xb4, 0x13, 0x24,
+	0xf5, 0xb1, 0x73, 0x58, 0xd2, 0x07, 0x54, 0xc1, 0x8f, 0x01, 0x17, 0x31, 0xd5, 0x1b, 0x54, 0x55,
+	0xf1, 0xb3, 0xac, 0x6a, 0x2a, 0x51, 0x1d, 0xd8, 0xe7, 0x92, 0x30, 0x1a, 0x2c, 0x51, 0x5d, 0x5d,
+	0x39, 0x5c, 0xc6, 0x92, 0xcd, 0x7a, 0x53, 0x1e, 0xb3, 0x00, 0x35, 0xf0, 0xa7, 0xb0, 0xa1, 0x7b,
+	0x7f, 0x38, 0x15, 0xca, 0x26, 0x5b, 0x76, 0x28, 0x50, 0xbc, 0x0c, 0x16, 0x72, 0x70, 0x41, 0x68,
+	0x34, 0x61, 0x48, 0x6d, 0x87, 0xb6, 0xcf, 0x65, 0x6e, 0x6b, 0xa2, 0x0b, 0xbc, 0x01, 0x68, 0x20,
+	0x2f, 0x99, 0x30, 0x51, 0x4d, 0xd1, 0x7f, 0xd5, 0x8e, 0xbc, 0x9b, 0x3f, 0xb7, 0xd6, 0xde, 0xde,
+	0x6e, 0x59, 0x37, 0xb7, 0x5b, 0xd6, 0x1f, 0xb7, 0x5b, 0xd6, 0xf7, 0x07, 0xb9, 0xef, 0xcb, 0x8c,
+	0x4a, 0x11, 0x5e, 0x73, 0x11, 0x4e, 0xc2, 0x28, 0x15, 0x22, 0xb6, 0x37, 0x7f, 0x33, 0xd9, 0x9b,
+	0x9f, 0xef, 0xad, 0x46, 0xe7, 0xbc, 0xaa, 0xff, 0x32, 0x07, 0xff, 0x06, 0x00, 0x00, 0xff, 0xff,
+	0x58, 0x2a, 0xfc, 0x07, 0x1a, 0x09, 0x00, 0x00,
 }
 
 func (m *LogShardInfo) Marshal() (dAtA []byte, err error) {
@@ -1280,7 +1325,12 @@ func (m *LogRecord) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		copy(dAtA[i:], m.Data)
 		i = encodeVarintLogservice(dAtA, i, uint64(len(m.Data)))
 		i--
-		dAtA[i] = 0x12
+		dAtA[i] = 0x1a
+	}
+	if m.Type != 0 {
+		i = encodeVarintLogservice(dAtA, i, uint64(m.Type))
+		i--
+		dAtA[i] = 0x10
 	}
 	if m.Index != 0 {
 		i = encodeVarintLogservice(dAtA, i, uint64(m.Index))
@@ -1641,6 +1691,9 @@ func (m *LogRecord) Size() (n int) {
 	if m.Index != 0 {
 		n += 1 + sovLogservice(uint64(m.Index))
 	}
+	if m.Type != 0 {
+		n += 1 + sovLogservice(uint64(m.Type))
+	}
 	l = len(m.Data)
 	if l > 0 {
 		n += 1 + l + sovLogservice(uint64(l))
@@ -2767,6 +2820,25 @@ func (m *LogRecord) Unmarshal(dAtA []byte) error {
 				}
 			}
 		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Type", wireType)
+			}
+			m.Type = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLogservice
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Type |= RecordType(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 3:
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field Data", wireType)
 			}
diff --git a/proto/logservice.proto b/proto/logservice.proto
index 21708e180..a95bc843e 100644
--- a/proto/logservice.proto
+++ b/proto/logservice.proto
@@ -17,6 +17,8 @@ option go_package = "github.com/matrixorigin/matrixone/pkg/pb/logservice";
 
 import "github.com/gogo/protobuf/gogoproto/gogo.proto";
 
+option (gogoproto.goproto_enum_prefix_all) = false;
+
 // LogShardInfo contains information a log shard.
 message LogShardInfo {
   // ShardID is the ID of a Log shard.
@@ -99,9 +101,18 @@ enum MethodType {
   CONNECT_RO   = 7;
 };
 
+enum RecordType {
+  UserRecord    = 0;
+  Internal      = 1;
+  LeaseUpdate   = 2;
+  LeaseRejected = 3;
+}
+
+// LogRecord is what we store into the LogService.
 message LogRecord {
-  uint64 Index = 1;
-  bytes Data   = 2;
+  uint64 Index    = 1;
+  RecordType Type = 2;
+  bytes Data      = 3;
 };
 
 message Request {


Test patch:
diff --git a/pkg/logservice/errors_test.go b/pkg/logservice/errors_test.go
index 76867b330..7a9351b87 100644
--- a/pkg/logservice/errors_test.go
+++ b/pkg/logservice/errors_test.go
@@ -42,7 +42,7 @@ func TestErrorConversion(t *testing.T) {
 func TestUnknownErrorIsHandled(t *testing.T) {
 	err := errors.New("test error")
 	code, str := toErrorCode(err)
-	assert.Equal(t, pb.ErrorCode_OtherSystemError, code)
+	assert.Equal(t, pb.OtherSystemError, code)
 	assert.Equal(t, err.Error(), str)
 }
 
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index 8e0a41547..d487682d5 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -69,13 +69,13 @@ func TestNewService(t *testing.T) {
 func TestServiceConnect(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		req := pb.Request{
-			Method:  pb.MethodType_CONNECT,
+			Method:  pb.CONNECT,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			DNID:    100,
 		}
 		resp := s.handleConnect(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 	}
 	runServiceTest(t, fn)
@@ -84,13 +84,13 @@ func TestServiceConnect(t *testing.T) {
 func TestServiceConnectTimeout(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		req := pb.Request{
-			Method:  pb.MethodType_CONNECT,
+			Method:  pb.CONNECT,
 			ShardID: 1,
 			Timeout: 50 * int64(time.Millisecond),
 			DNID:    100,
 		}
 		resp := s.handleConnect(req)
-		assert.Equal(t, pb.ErrorCode_Timeout, resp.ErrorCode)
+		assert.Equal(t, pb.Timeout, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 	}
 	runServiceTest(t, fn)
@@ -99,13 +99,13 @@ func TestServiceConnectTimeout(t *testing.T) {
 func TestServiceConnectRO(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		req := pb.Request{
-			Method:  pb.MethodType_CONNECT_RO,
+			Method:  pb.CONNECT_RO,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			DNID:    100,
 		}
 		resp := s.handleConnect(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 	}
 	runServiceTest(t, fn)
@@ -122,24 +122,24 @@ func getTestAppendCmd(id uint64, data []byte) []byte {
 func TestServiceHandleAppend(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		req := pb.Request{
-			Method:  pb.MethodType_CONNECT_RO,
+			Method:  pb.CONNECT_RO,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			DNID:    100,
 		}
 		resp := s.handleConnect(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 
 		data := make([]byte, 8)
 		cmd := getTestAppendCmd(req.DNID, data)
 		req = pb.Request{
-			Method:  pb.MethodType_APPEND,
+			Method:  pb.APPEND,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 		}
 		resp = s.handleAppend(req, cmd)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 		assert.Equal(t, uint64(4), resp.Index)
 	}
@@ -149,24 +149,24 @@ func TestServiceHandleAppend(t *testing.T) {
 func TestServiceHandleAppendWhenNotBeingTheLeaseHolder(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		req := pb.Request{
-			Method:  pb.MethodType_CONNECT_RO,
+			Method:  pb.CONNECT_RO,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			DNID:    100,
 		}
 		resp := s.handleConnect(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 
 		data := make([]byte, 8)
 		cmd := getTestAppendCmd(req.DNID+1, data)
 		req = pb.Request{
-			Method:  pb.MethodType_APPEND,
+			Method:  pb.APPEND,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 		}
 		resp = s.handleAppend(req, cmd)
-		assert.Equal(t, pb.ErrorCode_NotLeaseHolder, resp.ErrorCode)
+		assert.Equal(t, pb.NotLeaseHolder, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 		assert.Equal(t, uint64(0), resp.Index)
 	}
@@ -176,40 +176,44 @@ func TestServiceHandleAppendWhenNotBeingTheLeaseHolder(t *testing.T) {
 func TestServiceHandleRead(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		req := pb.Request{
-			Method:  pb.MethodType_CONNECT_RO,
+			Method:  pb.CONNECT_RO,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			DNID:    100,
 		}
 		resp := s.handleConnect(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 
 		data := make([]byte, 8)
 		cmd := getTestAppendCmd(req.DNID, data)
 		req = pb.Request{
-			Method:  pb.MethodType_APPEND,
+			Method:  pb.APPEND,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 		}
 		resp = s.handleAppend(req, cmd)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 		assert.Equal(t, uint64(4), resp.Index)
 
 		req = pb.Request{
-			Method:  pb.MethodType_READ,
+			Method:  pb.READ,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			Index:   1,
 			MaxSize: 1024 * 32,
 		}
 		resp, records := s.handleRead(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 		assert.Equal(t, uint64(1), resp.LastIndex)
-		require.Equal(t, 1, len(records.Records))
-		assert.Equal(t, cmd, records.Records[0].Data)
+		require.Equal(t, 4, len(records.Records))
+		assert.Equal(t, pb.Internal, records.Records[0].Type)
+		assert.Equal(t, pb.Internal, records.Records[1].Type)
+		assert.Equal(t, pb.LeaseUpdate, records.Records[2].Type)
+		assert.Equal(t, pb.UserRecord, records.Records[3].Type)
+		assert.Equal(t, cmd, records.Records[3].Data)
 	}
 	runServiceTest(t, fn)
 }
@@ -217,56 +221,56 @@ func TestServiceHandleRead(t *testing.T) {
 func TestServiceTruncate(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		req := pb.Request{
-			Method:  pb.MethodType_CONNECT_RO,
+			Method:  pb.CONNECT_RO,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			DNID:    100,
 		}
 		resp := s.handleConnect(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 
 		data := make([]byte, 8)
 		cmd := getTestAppendCmd(req.DNID, data)
 		req = pb.Request{
-			Method:  pb.MethodType_APPEND,
+			Method:  pb.APPEND,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 		}
 		resp = s.handleAppend(req, cmd)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 		assert.Equal(t, uint64(4), resp.Index)
 
 		req = pb.Request{
-			Method:  pb.MethodType_TRUNCATE,
+			Method:  pb.TRUNCATE,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			Index:   4,
 		}
 		resp = s.handleTruncate(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 		assert.Equal(t, uint64(0), resp.Index)
 
 		req = pb.Request{
-			Method:  pb.MethodType_GET_TRUNCATE,
+			Method:  pb.GET_TRUNCATE,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 		}
 		resp = s.handleGetTruncatedIndex(req)
-		assert.Equal(t, pb.ErrorCode_NoError, resp.ErrorCode)
+		assert.Equal(t, pb.NoError, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 		assert.Equal(t, uint64(4), resp.Index)
 
 		req = pb.Request{
-			Method:  pb.MethodType_TRUNCATE,
+			Method:  pb.TRUNCATE,
 			ShardID: 1,
 			Timeout: int64(time.Second),
 			Index:   3,
 		}
 		resp = s.handleTruncate(req)
-		assert.Equal(t, pb.ErrorCode_IndexAlreadyTruncated, resp.ErrorCode)
+		assert.Equal(t, pb.IndexAlreadyTruncated, resp.ErrorCode)
 		assert.Equal(t, "", resp.ErrorMessage)
 	}
 	runServiceTest(t, fn)
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index ff52c5bc9..90939d764 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -237,9 +237,12 @@ func TestQueryLog(t *testing.T) {
 		// lease holder ID update cmd at entry index 3
 		entries, lsn, err = store.QueryLog(ctx, 1, 3, math.MaxUint64)
 		assert.NoError(t, err)
-		assert.Equal(t, 1, len(entries))
+		assert.Equal(t, 2, len(entries))
 		assert.Equal(t, uint64(3), lsn)
-		assert.Equal(t, entries[0].Data, cmd)
+		assert.Equal(t, cmd, entries[1].Data)
+		assert.Equal(t, pb.LeaseUpdate, entries[0].Type)
+		assert.Equal(t, pb.UserRecord, entries[1].Type)
+
 		// size limited
 		_, err = store.Append(ctx, 1, cmd)
 		assert.NoError(t, err)
diff --git a/pkg/logservice/transport_test.go b/pkg/logservice/transport_test.go
index aa8103df2..653cdee2b 100644
--- a/pkg/logservice/transport_test.go
+++ b/pkg/logservice/transport_test.go
@@ -97,7 +97,7 @@ func TestReadSize(t *testing.T) {
 func TestWriteRequest(t *testing.T) {
 	conn := newTestConn()
 	req := pb.Request{
-		Method:      pb.MethodType_APPEND,
+		Method:      pb.APPEND,
 		Name:        "test",
 		DNID:        1234567890,
 		PayloadSize: 32,
@@ -123,7 +123,7 @@ func TestWriteRequest(t *testing.T) {
 func TestReadRequest(t *testing.T) {
 	conn := newTestConn()
 	req := pb.Request{
-		Method:      pb.MethodType_APPEND,
+		Method:      pb.APPEND,
 		Name:        "test",
 		PayloadSize: 32,
 	}
@@ -155,7 +155,7 @@ func TestWriteResponse(t *testing.T) {
 func testWriteResponse(t *testing.T, sz int) {
 	conn := newTestConn()
 	resp := pb.Response{
-		Method:    pb.MethodType_APPEND,
+		Method:    pb.APPEND,
 		ShardID:   1234567890,
 		LastIndex: 234567890,
 	}
@@ -206,7 +206,7 @@ func TestReadResponse(t *testing.T) {
 func testReadResponse(t *testing.T, sz int) {
 	conn := newTestConn()
 	resp := pb.Response{
-		Method:    pb.MethodType_APPEND,
+		Method:    pb.APPEND,
 		ShardID:   1234567890,
 		LastIndex: 234567890,
 	}