dn: add dn store (#4099)

pkg/txn/clock/hlc.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/matrixorigin/matrixone/pkg/common/stopper"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
 )
 
@@ -89,6 +90,19 @@ func NewUnixNanoHLCClock(ctx context.Context, maxOffset time.Duration) *HLCClock
 	return clock
 }
 
+// NewUnixNanoHLCClockWithStopper is similar to NewUnixNanoHLCClock, but perform
+// clock check use stopper
+func NewUnixNanoHLCClockWithStopper(stopper *stopper.Stopper, maxOffset time.Duration) *HLCClock {
+	clock := &HLCClock{
+		physicalClock: physicalClock,
+		maxOffset:     maxOffset,
+	}
+	if err := stopper.RunTask(clock.offsetMonitor); err != nil {
+		panic(err)
+	}
+	return clock
+}
+
 // HasNetworkLatency returns a boolean value indicating whether there is network
 // latency involved when retrieving timestamps. There is no such network latency
 // in HLCClock.

pkg/txn/service/service.go

@@ -31,6 +31,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/txn/rpc"
 	"github.com/matrixorigin/matrixone/pkg/txn/storage"
 	"github.com/matrixorigin/matrixone/pkg/txn/util"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 )
 
@@ -108,6 +109,9 @@ func (s *service) Start() error {
 func (s *service) Close() error {
 	s.waitRecoveryCompleted()
 	s.stopper.Stop()
+	if err := s.storage.Close(); err != nil {
+		return multierr.Append(err, s.sender.Close())
+	}
 	return s.sender.Close()
 }
 

pkg/txn/storage/mem/kv_txn_storage.go

@@ -183,6 +183,10 @@ func (kv *KVTxnStorage) StartRecovery(c chan txn.TxnMeta) {
 	}
 }
 
+func (kv *KVTxnStorage) Close() error {
+	return nil
+}
+
 func (kv *KVTxnStorage) GetLatestLsn() logservice.Lsn {
 	return kv.latest
 }

pkg/txn/storage/types.go

@@ -34,6 +34,8 @@ type TxnStorage interface {
 	// StartRecovery start txnStorage recovery process. Use the incoming channel to send the Txn that needs to be
 	// recovered and close the channel when all the logs have been processed.
 	StartRecovery(chan txn.TxnMeta)
+	// Close close the txn storage
+	Close() error
 
 	// Read execute read requests sent by CN.
 	//

proto/metadata.proto

@@ -50,3 +50,11 @@ message LogShardRecord {
   // Name is the human readable name of the shard given by the DN.
   string Name = 3;
 };
+
+// DNStore DN store metadata
+message DNStore {
+  // UUID DNStore uuid id
+  string UUID = 1;
+  // Shards DNShards
+  repeated DNShard Shards = 2 [(gogoproto.nullable) = false];
+}
\ No newline at end of file