Refactor snapshotTS to include last commit TS

Author: bootjp

adapter/dynamodb.go

@@ -114,7 +114,7 @@ func (d *DynamoDBServer) getItem(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "missing key", http.StatusBadRequest)
 		return
 	}
-	readTS := snapshotTS(d.coordinator.Clock())
+	readTS := snapshotTS(d.coordinator.Clock(), d.store)
 	v, err := d.store.GetAt(r.Context(), []byte(keyAttr.S), readTS)
 	if err != nil {
 		if errors.Is(err, store.ErrKeyNotFound) {
@@ -234,7 +234,7 @@ func (d *DynamoDBServer) validateCondition(ctx context.Context, expr string, nam
 	if expr == "" {
 		return nil
 	}
-	readTS := snapshotTS(d.coordinator.Clock())
+	readTS := snapshotTS(d.coordinator.Clock(), d.store)
 	exists, err := d.store.ExistsAt(ctx, key, readTS)
 	if err != nil {
 		return errors.WithStack(err)

adapter/grpc.go

@@ -42,7 +42,7 @@ func NewGRPCServer(store store.MVCCStore, coordinate *kv.Coordinate) *GRPCServer
 func (r GRPCServer) RawGet(ctx context.Context, req *pb.RawGetRequest) (*pb.RawGetResponse, error) {
 	readTS := req.GetTs()
 	if readTS == 0 {
-		readTS = snapshotTS(r.coordinator.Clock())
+		readTS = snapshotTS(r.coordinator.Clock(), r.store)
 	}
 
 	if r.coordinator.IsLeader() {
@@ -98,7 +98,7 @@ func (r GRPCServer) tryLeaderGet(key []byte) ([]byte, error) {
 	defer conn.Close()
 
 	cli := pb.NewRawKVClient(conn)
-	ts := snapshotTS(r.coordinator.Clock())
+	ts := snapshotTS(r.coordinator.Clock(), r.store)
 	resp, err := cli.RawGet(context.Background(), &pb.RawGetRequest{Key: key, Ts: ts})
 	if err != nil {
 		return nil, errors.WithStack(err)
@@ -186,7 +186,7 @@ func (r GRPCServer) Get(ctx context.Context, req *pb.GetRequest) (*pb.GetRespons
 		return nil, errors.WithStack(err)
 	}
 
-	readTS := snapshotTS(r.coordinator.Clock())
+	readTS := snapshotTS(r.coordinator.Clock(), r.store)
 	v, err := r.store.GetAt(ctx, req.Key, readTS)
 	if err != nil {
 		switch {
@@ -234,7 +234,7 @@ func (r GRPCServer) Scan(ctx context.Context, req *pb.ScanRequest) (*pb.ScanResp
 			Kv: nil,
 		}, errors.WithStack(err)
 	}
-	readTS := snapshotTS(r.coordinator.Clock())
+	readTS := snapshotTS(r.coordinator.Clock(), r.store)
 	res, err := r.store.ScanAt(ctx, req.StartKey, req.EndKey, limit, readTS)
 	if err != nil {
 		return &pb.ScanResponse{

adapter/redis.go

@@ -109,7 +109,7 @@ func getConnState(conn redcon.Conn) *connState {
 }
 
 func (r *RedisServer) readTS() uint64 {
-	return snapshotTS(r.coordinator.Clock())
+	return snapshotTS(r.coordinator.Clock(), r.store)
 }
 
 func (r *RedisServer) Run() error {

adapter/ts.go

@@ -1,17 +1,26 @@
 package adapter
 
-import "github.com/bootjp/elastickv/kv"
+import (
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+)
 
-// snapshotTS returns a timestamp suitable for snapshot reads without
-// unnecessarily advancing the logical clock. It relies solely on the shared
-// HLC; if none has been issued yet, fall back to MaxUint64 to see latest
-// committed versions irrespective of local clock lag.
-func snapshotTS(clock *kv.HLC) uint64 {
-	if clock == nil {
-		return ^uint64(0)
+// snapshotTS picks a safe snapshot timestamp:
+// - uses the store's last commit watermark if available,
+// - otherwise the coordinator's HLC current value,
+// - and falls back to MaxUint64 if neither is set.
+func snapshotTS(clock *kv.HLC, st store.MVCCStore) uint64 {
+	ts := uint64(0)
+	if st != nil {
+		ts = st.LastCommitTS()
 	}
-	if cur := clock.Current(); cur != 0 {
-		return cur
+	if clock != nil {
+		if cur := clock.Current(); cur > ts {
+			ts = cur
+		}
 	}
-	return ^uint64(0)
+	if ts == 0 {
+		ts = ^uint64(0)
+	}
+	return ts
 }

store/mvcc_store.go

@@ -61,6 +61,14 @@ type mvccStore struct {
 	lastCommitTS uint64
 }
 
+// LastCommitTS exposes the latest commit timestamp for read snapshot selection.
+// It is intentionally not part of the public MVCCStore interface.
+func (s *mvccStore) LastCommitTS() uint64 {
+	s.mtx.RLock()
+	defer s.mtx.RUnlock()
+	return s.lastCommitTS
+}
+
 // NewMVCCStore creates a new MVCC-enabled in-memory store.
 func NewMVCCStore() MVCCStore {
 	return &mvccStore{

store/store.go

@@ -59,6 +59,8 @@ type MVCCStore interface {
 	// It must return ErrWriteConflict if any key has a newer commit timestamp
 	// than startTS.
 	ApplyMutations(ctx context.Context, mutations []*KVPairMutation, startTS, commitTS uint64) error
+	// LastCommitTS returns the highest commit timestamp applied on this node.
+	LastCommitTS() uint64
 	Snapshot() (io.ReadWriter, error)
 	Restore(buf io.Reader) error
 	Close() error