Merge pull request #274 from bootjp/feature/fix-create-node

Removed unnecessary waitApply and fixed a bug that caused stale reads…

Author: bootjp

adapter/grpc.go

@@ -11,6 +11,8 @@ import (
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
 	"github.com/spaolacci/murmur3"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 var _ pb.RawKVServer = (*GRPCServer)(nil)
@@ -38,16 +40,32 @@ func NewGRPCServer(store store.ScanStore, coordinate *kv.Coordinate) *GRPCServer
 }
 
 func (r GRPCServer) RawGet(ctx context.Context, req *pb.RawGetRequest) (*pb.RawGetResponse, error) {
-	v, err := r.store.Get(ctx, req.Key)
-	if err != nil {
-		switch {
-		case errors.Is(err, store.ErrKeyNotFound):
-			return &pb.RawGetResponse{
-				Value: nil,
-			}, nil
-		default:
-			return nil, errors.WithStack(err)
+	if r.coordinator.IsLeader() {
+		v, err := r.store.Get(ctx, req.Key)
+		if err != nil {
+			switch {
+			case errors.Is(err, store.ErrKeyNotFound):
+				return &pb.RawGetResponse{
+					Value: nil,
+				}, nil
+			default:
+				return nil, errors.WithStack(err)
+			}
 		}
+		r.log.InfoContext(ctx, "Get",
+			slog.String("key", string(req.Key)),
+			slog.String("value", string(v)))
+
+		return &pb.RawGetResponse{
+			Value: v,
+		}, nil
+	}
+
+	v, err := r.tryLeaderGet(req.Key)
+	if err != nil {
+		return &pb.RawGetResponse{
+			Value: nil,
+		}, err
 	}
 
 	r.log.InfoContext(ctx, "Get",
@@ -59,6 +77,30 @@ func (r GRPCServer) RawGet(ctx context.Context, req *pb.RawGetRequest) (*pb.RawG
 	}, nil
 }
 
+func (r GRPCServer) tryLeaderGet(key []byte) ([]byte, error) {
+	addr := r.coordinator.RaftLeader()
+	if addr == "" {
+		return nil, ErrLeaderNotFound
+	}
+
+	conn, err := grpc.NewClient(string(addr),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultCallOptions(grpc.WaitForReady(true)),
+	)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	defer conn.Close()
+
+	cli := pb.NewRawKVClient(conn)
+	resp, err := cli.RawGet(context.Background(), &pb.RawGetRequest{Key: key})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return resp.Value, nil
+}
+
 func (r GRPCServer) RawPut(_ context.Context, req *pb.RawPutRequest) (*pb.RawPutResponse, error) {
 	m, err := r.grpcTranscoder.RawPutToRequest(req)
 	if err != nil {

adapter/internal.go

@@ -26,6 +26,7 @@ type Internal struct {
 var _ pb.InternalServer = (*Internal)(nil)
 
 var ErrNotLeader = errors.New("not leader")
+var ErrLeaderNotFound = errors.New("leader not found")
 
 func (i *Internal) Forward(_ context.Context, req *pb.ForwardRequest) (*pb.ForwardResponse, error) {
 	if i.raft.State() != raft.Leader {

adapter/redis.go

@@ -116,18 +116,29 @@ func (r *RedisServer) set(conn redcon.Conn, cmd redcon.Command) {
 }
 
 func (r *RedisServer) get(conn redcon.Conn, cmd redcon.Command) {
-	if v, ok := r.tryLeaderGet(cmd.Args[1]); ok {
-		if v == nil {
-			conn.WriteNull()
-		} else {
-			conn.WriteBulk(v)
+	if r.coordinator.IsLeader() {
+		v, err := r.store.Get(context.Background(), cmd.Args[1])
+		if err != nil {
+			switch {
+			case errors.Is(err, store.ErrKeyNotFound):
+				conn.WriteNull()
+			default:
+				conn.WriteError(err.Error())
+			}
+			return
 		}
+		conn.WriteBulk(v)
 		return
 	}
 
-	v, err := r.store.Get(context.Background(), cmd.Args[1])
+	v, err := r.tryLeaderGet(cmd.Args[1])
 	if err != nil {
-		conn.WriteNull()
+		switch {
+		case errors.Is(err, store.ErrKeyNotFound):
+			conn.WriteNull()
+		default:
+			conn.WriteError(err.Error())
+		}
 		return
 	}
 
@@ -205,29 +216,26 @@ func (r *RedisServer) keys(conn redcon.Conn, cmd redcon.Command) {
 
 // tryLeaderGet proxies a GET to the current Raft leader, returning the value and
 // whether the proxy succeeded.
-func (r *RedisServer) tryLeaderGet(key []byte) ([]byte, bool) {
-	c, ok := r.coordinator.(*kv.Coordinate)
-	if !ok || r.coordinator.IsLeader() {
-		return nil, false
-	}
-	addr := c.RaftLeader()
+func (r *RedisServer) tryLeaderGet(key []byte) ([]byte, error) {
+	addr := r.coordinator.RaftLeader()
 	if addr == "" {
-		return nil, false
+		return nil, ErrLeaderNotFound
 	}
 
 	conn, err := grpc.NewClient(string(addr),
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithDefaultCallOptions(grpc.WaitForReady(true)),
 	)
 	if err != nil {
-		return nil, false
+		return nil, errors.WithStack(err)
 	}
 	defer conn.Close()
 
 	cli := pb.NewRawKVClient(conn)
 	resp, err := cli.RawGet(context.Background(), &pb.RawGetRequest{Key: key})
 	if err != nil {
-		return nil, false
+		return nil, errors.WithStack(err)
 	}
-	return resp.Value, true
+
+	return resp.Value, nil
 }

adapter/redis_test.go

@@ -68,7 +68,7 @@ func TestRedis_follower_redirect_node_set_get_deleted(t *testing.T) {
 	assert.Equal(t, int64(1), res3.Val())
 
 	res4 := rdb.Get(ctx, string(key))
-	assert.Equal(t, redis.Nil, res4.Err())
+	assert.NoError(t, res4.Err())
 	assert.Equal(t, "", res4.Val())
 }
 

kv/coordinator.go

@@ -2,7 +2,6 @@ package kv
 
 import (
 	"context"
-	"time"
 
 	pb "github.com/bootjp/elastickv/proto"
 	"github.com/cockroachdb/errors"
@@ -32,6 +31,7 @@ var _ Coordinator = (*Coordinate)(nil)
 type Coordinator interface {
 	Dispatch(reqs *OperationGroup[OP]) (*CoordinateResponse, error)
 	IsLeader() bool
+	RaftLeader() raft.ServerAddress
 }
 
 func (c *Coordinate) Dispatch(reqs *OperationGroup[OP]) (*CoordinateResponse, error) {
@@ -209,41 +209,11 @@ func (c *Coordinate) redirect(reqs *OperationGroup[OP]) (*CoordinateResponse, er
 		return nil, ErrInvalidRequest
 	}
 
-	if err := c.waitApplied(r.CommitIndex); err != nil {
-		return nil, errors.WithStack(err)
-	}
-
 	return &CoordinateResponse{
 		CommitIndex: r.CommitIndex,
 	}, nil
 }
 
-// waitApplied blocks until the local raft FSM has applied at least the given
-// log index. This is used on followers after redirecting a write to the leader
-// so that immediately subsequent reads served locally observe the new value.
-func (c *Coordinate) waitApplied(idx uint64) error {
-	// Fast path: already applied.
-	if c.raft.AppliedIndex() >= idx {
-		return nil
-	}
-
-	const (
-		pollInterval = 5 * time.Millisecond
-		// Keep generous but finite to avoid hanging client calls.
-		timeout = 2 * time.Second
-	)
-
-	deadline := time.Now().Add(timeout)
-	for time.Now().Before(deadline) {
-		if c.raft.AppliedIndex() >= idx {
-			return nil
-		}
-		time.Sleep(pollInterval)
-	}
-
-	return errors.Wrapf(ErrInvalidRequest, "timeout waiting for apply: want=%d applied=%d", idx, c.raft.AppliedIndex())
-}
-
 func (c *Coordinate) toForwardRequest(reqs []*pb.Request) *pb.ForwardRequest {
 	if len(reqs) == 0 {
 		return nil