Add check for mod revision=0 in update txn

pkg/server/delete.go

@@ -42,6 +42,7 @@ func (l *LimitedServer) delete(ctx context.Context, key string, revision int64)
 						ResponseRange: &etcdserverpb.RangeResponse{
 							Header: txnHeader(rev),
 							Kvs:    toKVs(kv),
+							Count:  1,
 						},
 					},
 				},

pkg/server/kv.go

@@ -2,7 +2,6 @@ package server
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/sirupsen/logrus"
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
@@ -94,17 +93,17 @@ func toKV(kv *KeyValue) *mvccpb.KeyValue {
 }
 
 func (k *KVServerBridge) Put(ctx context.Context, r *etcdserverpb.PutRequest) (*etcdserverpb.PutResponse, error) {
-	return nil, fmt.Errorf("put is not supported")
+	return nil, unsupported("put")
 }
 
 func (k *KVServerBridge) DeleteRange(ctx context.Context, r *etcdserverpb.DeleteRangeRequest) (*etcdserverpb.DeleteRangeResponse, error) {
-	return nil, fmt.Errorf("delete is not supported")
+	return nil, unsupported("delete")
 }
 
 func (k *KVServerBridge) Txn(ctx context.Context, r *etcdserverpb.TxnRequest) (*etcdserverpb.TxnResponse, error) {
 	res, err := k.limited.Txn(ctx, r)
 	if err != nil {
-		logrus.Errorf("error in txn: %v", err)
+		logrus.Errorf("error in txn %s: %v", r, err)
 	}
 	return res, err
 }
@@ -116,7 +115,3 @@ func (k *KVServerBridge) Compact(ctx context.Context, r *etcdserverpb.Compaction
 		},
 	}, nil
 }
-
-func unsupported(field string) error {
-	return fmt.Errorf("%s is unsupported", field)
-}

pkg/server/limited.go

@@ -2,7 +2,6 @@ package server
 
 import (
 	"context"
-	"fmt"
 
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
@@ -38,7 +37,7 @@ func (l *LimitedServer) Txn(ctx context.Context, txn *etcdserverpb.TxnRequest) (
 	if isCompact(txn) {
 		return l.compact(ctx)
 	}
-	return nil, fmt.Errorf("unsupported transaction: %v", txn)
+	return nil, ErrNotSupported
 }
 
 type ResponseHeader struct {

pkg/server/types.go

@@ -5,9 +5,13 @@ import (
 	"database/sql"
 
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 var (
+	ErrNotSupported = status.New(codes.InvalidArgument, "etcdserver: unsupported operations in txn request").Err()
+
 	ErrKeyExists = rpctypes.ErrGRPCDuplicateKey
 	ErrCompacted = rpctypes.ErrGRPCCompacted
 )
@@ -70,3 +74,7 @@ type Event struct {
 	KV     *KeyValue
 	PrevKV *KeyValue
 }
+
+func unsupported(field string) error {
+	return status.New(codes.Unimplemented, field+" is not implemented by kine").Err()
+}

pkg/server/update.go

@@ -32,7 +32,11 @@ func (l *LimitedServer) update(ctx context.Context, rev int64, key string, value
 
 	if rev == 0 {
 		rev, err = l.backend.Create(ctx, key, value, lease)
-		ok = true
+		if err == ErrKeyExists {
+			rev, kv, err = l.backend.Get(ctx, key, "", 1, rev)
+		} else {
+			ok = true
+		}
 	} else {
 		rev, kv, ok, err = l.backend.Update(ctx, key, value, rev, lease)
 	}
@@ -62,6 +66,7 @@ func (l *LimitedServer) update(ctx context.Context, rev int64, key string, value
 					ResponseRange: &etcdserverpb.RangeResponse{
 						Header: txnHeader(rev),
 						Kvs:    toKVs(kv),
+						Count:  1,
 					},
 				},
 			},