allow deletion of non-exist key (#220)

We should allow deletions of non-exist keys, which is the behavior of
old versions of birdwatcher.
Let's keep the behavior consistent.

states/kv/kv.go

@@ -133,7 +133,8 @@ func (kv *etcdKV) removeWithPrevKV(ctx context.Context, key string) (*mvccpb.Key
 	if len(resp.PrevKvs) > 0 {
 		return resp.PrevKvs[0], nil
 	}
-	return nil, fmt.Errorf("Error getting prev kv in removeWithPrevKV for key: %s", key)
+	// Note: we allow response to be empty, which suggests that the key doesn't exist e.g. it's already deleted.
+	return nil, nil
 }
 
 func (kv *etcdKV) removeWithPrefixAndPrevKV(ctx context.Context, prefix string) ([]*mvccpb.KeyValue, error) {
@@ -463,7 +464,8 @@ func (kv *txnTiKV) RemoveWithPrefix(ctx context.Context, prefix string) error {
 func (kv *txnTiKV) removeWithPrevKV(ctx context.Context, key string) (*mvccpb.KeyValue, error) {
 	preV, err := kv.Load(ctx, key)
 	if err != nil {
-		return nil, err
+		// Note: we allow response to be empty, which suggests that the key doesn't exist e.g. it's already deleted.
+		return nil, nil
 	}
 	err = kv.Remove(ctx, key)
 	if err != nil {

states/kv/kv_audit.go

@@ -102,6 +102,9 @@ func (c *FileAuditKV) writeHeader(op models.AuditOpType, entriesNum int32) {
 }
 
 func (c *FileAuditKV) writeLogKV(kv *mvccpb.KeyValue) {
+	if kv == (*mvccpb.KeyValue)(nil) {
+		return
+	}
 	bs, _ := proto.Marshal(kv)
 	c.writeData(bs)
 }

states/kv/txn_tikv_test.go → states/kv/kv_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 )
 
 func TestTiKVLoad(te *testing.T) {
@@ -139,6 +140,9 @@ func TestTiKVLoad(te *testing.T) {
 			assert.NoError(t, err)
 			assert.ElementsMatch(t, []string{"testr2", "testr2/c"}, keys)
 			assert.ElementsMatch(t, []string{"value2", "value3"}, vals)
+			// allow remove with non-exist prefix
+			err = kv.RemoveWithPrefix(ctx, "testnoexist")
+			assert.NoError(t, err)
 		}
 	})
 }
@@ -224,14 +228,22 @@ func TestRemoveWithPrev(t *testing.T) {
 			err := kv.Save(ctx, test.key, test.value)
 			assert.NoError(t, err)
 		}
-		kvp, err := kv.removeWithPrevKV(ctx, "testr1")
+		// it's fine if the key doesn't exist.
+		kvp, err := kv.removeWithPrevKV(ctx, "not_exist_key")
+		assert.NoError(t, err)
+		assert.Equal(t, kvp, (*mvccpb.KeyValue)(nil))
+		kvp, err = kv.removeWithPrevKV(ctx, "testr1")
 		assert.NoError(t, err)
 		assert.Equal(t, string(kvp.Key), "testr1")
 		assert.Equal(t, string(kvp.Value), "value1")
 		keys, vals, err := kv.LoadWithPrefix(ctx, "testr")
 		assert.NoError(t, err)
 		assert.ElementsMatch(t, []string{"testr2", "testr1/a", "testr1/b", "testr2/c"}, keys)
 		assert.ElementsMatch(t, []string{"value2", "value_a", "value_b", "value3"}, vals)
+		// it's fine if the key doesn't exist.
+		kvp, err = kv.removeWithPrevKV(ctx, "testr1")
+		assert.NoError(t, err)
+		assert.Equal(t, kvp, (*mvccpb.KeyValue)(nil))
 	}
 }