Merge main branch

Author: bootjp

.github/workflows/docker-image.yml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         with:
           submodules: true
 #      - name: Set up QEMU

.github/workflows/go-mod-fix.yaml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           fetch-depth: 2
       - name: fix

.github/workflows/go-test.yml

@@ -15,7 +15,7 @@ jobs:
         os: [ [ubuntu-latest] ]
     runs-on: ${{ matrix.os }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
       - uses: actions/setup-go@v5
         with:
           go-version-file: 'go.mod'

.github/workflows/golangci-lint.yml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Check out code into the Go module directory
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
       - name: golangci-lint
         uses: reviewdog/action-golangci-lint@v2
         with:

.gitignore

@@ -19,3 +19,6 @@
 
 # Go workspace file
 go.work
+
+# Built binary
+elastickv

adapter/distribution_server.go

@@ -0,0 +1,43 @@
+package adapter
+
+import (
+	"context"
+
+	"github.com/bootjp/elastickv/distribution"
+	pb "github.com/bootjp/elastickv/proto"
+)
+
+// DistributionServer serves distribution related gRPC APIs.
+type DistributionServer struct {
+	engine *distribution.Engine
+	pb.UnimplementedDistributionServer
+}
+
+// NewDistributionServer creates a new server.
+func NewDistributionServer(e *distribution.Engine) *DistributionServer {
+	return &DistributionServer{engine: e}
+}
+
+// UpdateRoute allows updating route information.
+func (s *DistributionServer) UpdateRoute(start, end []byte, group uint64) {
+	s.engine.UpdateRoute(start, end, group)
+}
+
+// GetRoute returns route for a key.
+func (s *DistributionServer) GetRoute(ctx context.Context, req *pb.GetRouteRequest) (*pb.GetRouteResponse, error) {
+	r, ok := s.engine.GetRoute(req.Key)
+	if !ok {
+		return &pb.GetRouteResponse{}, nil
+	}
+	return &pb.GetRouteResponse{
+		Start:       r.Start,
+		End:         r.End,
+		RaftGroupId: r.GroupID,
+	}, nil
+}
+
+// GetTimestamp returns monotonically increasing timestamp.
+func (s *DistributionServer) GetTimestamp(ctx context.Context, req *pb.GetTimestampRequest) (*pb.GetTimestampResponse, error) {
+	ts := s.engine.NextTimestamp()
+	return &pb.GetTimestampResponse{Timestamp: ts}, nil
+}

adapter/dynamodb.go

@@ -33,18 +33,19 @@ type DynamoDBServer struct {
 }
 
 func NewDynamoDBServer(listen net.Listener, st store.ScanStore, coordinate *kv.Coordinate) *DynamoDBServer {
-	return &DynamoDBServer{
+	d := &DynamoDBServer{
 		listen:           listen,
 		store:            st,
 		coordinator:      coordinate,
 		dynamoTranscoder: newDynamoDBTranscoder(),
 	}
-}
-
-func (d *DynamoDBServer) Run() error {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", d.handle)
 	d.httpServer = &http.Server{Handler: mux, ReadHeaderTimeout: time.Second}
+	return d
+}
+
+func (d *DynamoDBServer) Run() error {
 	if err := d.httpServer.Serve(d.listen); err != nil && !errors.Is(err, http.ErrServerClosed) {
 		return errors.WithStack(err)
 	}