allow concurrent delete

Author: aep

kv/kv_test.go

@@ -38,6 +38,89 @@ func TestKVDoesPreventOutdatedWrites(t *testing.T) {
 	require.NoError(t, err, "w4 must succeed because it is fresh")
 }
 
+// this behaviour actually sucks because it means Delete in apogy must be locked
+// this test is just here in case they change their mind about it
+func TestKVDoesPreventOutdatedDoubleDelete(t *testing.T) {
+
+	k, err := NewTikv()
+	require.NoError(t, err)
+	defer k.Close()
+
+	var ctx = t.Context()
+	var key = []byte("TestKVDoesNotPreventOutdatedDoubleDelete")
+
+	prep := k.Write()
+	prep.Put(key, []byte("1"))
+	err = prep.Commit(ctx)
+	require.NoError(t, err)
+
+	w1 := k.Write()
+	w2 := k.Write()
+
+	w1.Del(key)
+	w2.Del(key)
+
+	err = w1.Commit(context.Background())
+	require.NoError(t, err)
+
+	err = w2.Commit(context.Background())
+	require.Error(t, err)
+}
+
+func TestKVDeleteAndPutMustConflict1(t *testing.T) {
+
+	k, err := NewTikv()
+	require.NoError(t, err)
+	defer k.Close()
+
+	var ctx = t.Context()
+	var key = []byte("TestKVDoesNotPreventOutdatedDoubleDelete")
+
+	prep := k.Write()
+	prep.Put(key, []byte("1"))
+	err = prep.Commit(ctx)
+	require.NoError(t, err)
+
+	w1 := k.Write()
+	w2 := k.Write()
+
+	w1.Put(key, []byte("2"))
+	w2.Del(key)
+
+	err = w1.Commit(context.Background())
+	require.NoError(t, err)
+
+	err = w2.Commit(context.Background())
+	require.Error(t, err)
+}
+
+func TestKVDeleteAndPutMustConflict2(t *testing.T) {
+
+	k, err := NewTikv()
+	require.NoError(t, err)
+	defer k.Close()
+
+	var ctx = t.Context()
+	var key = []byte("TestKVDoesNotPreventOutdatedDoubleDelete")
+
+	prep := k.Write()
+	prep.Put(key, []byte("1"))
+	err = prep.Commit(ctx)
+	require.NoError(t, err)
+
+	w1 := k.Write()
+	w2 := k.Write()
+
+	w1.Del(key)
+	w2.Put(key, []byte("2"))
+
+	err = w1.Commit(context.Background())
+	require.NoError(t, err)
+
+	err = w2.Commit(context.Background())
+	require.Error(t, err)
+}
+
 func TestKVDoesNotPreventDup(t *testing.T) {
 	k, err := NewTikv()
 	require.NoError(t, err)

kv/tikv.go

@@ -123,7 +123,11 @@ func (w *TikvWrite) Del(key []byte) error {
 	if w.err != nil {
 		return w.err
 	}
-	return w.txn.Delete(key)
+	err := w.txn.Delete(key)
+	if err != nil {
+		w.err = err
+	}
+	return err
 }
 
 func (r *TikvWrite) Iter(ctx context.Context, start []byte, end []byte) iter.Seq2[KeyAndValue, error] {

server/crud.go

@@ -71,6 +71,8 @@ func (s *server) PutDocument(c echo.Context) error {
 		hotKeys = append(hotKeys, path)
 	}
 
+	// TODO add unique
+
 	var w2 kv.Write
 	if len(hotKeys) == 0 {
 		w2 = s.kv.Write()
@@ -306,7 +308,7 @@ func (s *server) DeleteDocument(c echo.Context, model string, id string) error {
 		return echo.NewHTTPError(http.StatusBadRequest, err.Error())
 	}
 
-	w := s.kv.Write()
+	w, err := s.kv.ExclusiveWrite(ctx, path)
 	defer w.Close()
 
 	// First get the document to remove its indexes
@@ -315,12 +317,12 @@ func (s *server) DeleteDocument(c echo.Context, model string, id string) error {
 	if err != nil {
 		span.RecordError(err)
 		if tikerr.IsErrNotFound(err) {
-			return echo.NewHTTPError(http.StatusNotFound, "document not found")
+			return c.NoContent(http.StatusOK)
 		}
 		return echo.NewHTTPError(http.StatusInternalServerError, fmt.Sprintf("database error: %v", err))
 	}
 	if bytes == nil {
-		notFoundErr := echo.NewHTTPError(http.StatusNotFound, "document not found")
+		notFoundErr := echo.NewHTTPError(http.StatusNotFound, "document is empty")
 		span.RecordError(notFoundErr)
 		return notFoundErr
 	}

server/crud_test.go

@@ -483,3 +483,47 @@ func TestConcurrentMutations_NeverFail(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, int64(concurrentCount), counterInt, "All mutations should have been applied")
 }
+
+func TestConcurrentDelete(t *testing.T) {
+	e, s := setupTestServer(t)
+
+	// Create a document with a counter
+	docId := "concurrent-delete-test"
+	initialVal := map[string]interface{}{
+		"counter": json.Number("0"),
+	}
+	initialDoc := openapi.Document{
+		Model: "Test.com.example",
+		Id:    docId,
+		Val:   initialVal,
+	}
+
+	// First PUT to create the document
+	docBytes, _ := json.Marshal(initialDoc)
+	req := httptest.NewRequest(http.MethodPost, "/v1/", bytes.NewReader(docBytes))
+	req.Header.Set(echo.HeaderContentType, "application/json")
+	rec := httptest.NewRecorder()
+	c := e.NewContext(req, rec)
+
+	assert.NoError(t, s.PutDocument(c))
+
+	// Number of concurrent mutations to perform
+	concurrentCount := 50
+
+	// Launch concurrent mutations, all deleting the doc
+	var wg sync.WaitGroup
+	for i := 0; i < concurrentCount; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			reqMut := httptest.NewRequest(http.MethodDelete, "/v1", nil)
+			reqMut.Header.Set(echo.HeaderContentType, "application/json")
+
+			err := s.DeleteDocument(c, initialDoc.Model, initialDoc.Id)
+			assert.NoError(t, err, "Concurrent delete should not fail")
+		}()
+	}
+
+	wg.Wait()
+}