fix: race condition

Makefile

@@ -5,7 +5,10 @@ FLAGS = -ldflags "\
 "
 
 test:
-	go test -cover ./...
+	go test ./...
+
+cover:
+	GOTOOLCHAIN=go1.25.0+auto go test -cover ./...
 
 run:
 	STATICS=statics/www/ go run $(FLAGS) ./cmd/inceptiondb/...
@@ -52,3 +55,7 @@ book:
 .PHONY: version
 version:
 	@echo $(VERSION)
+
+.PHONY: bench
+bench:
+	go test -bench=BenchmarkIndexMap_RemoveRow_Concurrent ./collection/...

api/apicollectionv1/0_traverse.go

@@ -28,7 +28,7 @@ func traverse(requestBody []byte, col *collection.Collection, f func(row *collec
 		return err
 	}
 
-	hasFilter := options.Filter != nil && len(options.Filter) > 0
+	hasFilter := len(options.Filter) > 0
 
 	skip := options.Skip
 	limit := options.Limit

api/apicollectionv1/patch.go

@@ -6,7 +6,6 @@ import (
 	"io"
 	"net/http"
 
-	"github.com/SierraSoftworks/connor"
 	"github.com/fulldump/box"
 
 	"github.com/fulldump/inceptiondb/collection"
@@ -36,33 +35,15 @@ func patch(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
 
 	traverse(requestBody, col, func(row *collection.Row) bool {
 
-		row.PatchMutex.Lock()
-		defer row.PatchMutex.Unlock()
-
-		hasFilter := patch.Filter != nil && len(patch.Filter) > 0
-		if hasFilter {
-
-			rowData := map[string]interface{}{}
-			json.Unmarshal(row.Payload, &rowData) // todo: handle error here?
-
-			match, err := connor.Match(patch.Filter, rowData)
-			if err != nil {
-				// todo: handle error?
-				// return fmt.Errorf("match: %w", err)
-				return false
-			}
-			if !match {
-				return false
-			}
-		}
-
 		err := col.Patch(row, patch.Patch)
 		if err != nil {
 			// TODO: handle err??
 			// return err
 			return true // todo: OR return false?
 		}
 
+		row.PatchMutex.Lock()
+		defer row.PatchMutex.Unlock()
 		e.Encode(row.Payload) // todo: handle err?
 
 		return true

cmd/bench/main.go

@@ -27,7 +27,7 @@ func main() {
 	}()
 
 	c := Config{
-		Test:    "patch",
+		Test:    "remove",
 		Base:    "",
 		N:       1_000_000,
 		Workers: 16,
@@ -40,6 +40,8 @@ func main() {
 		TestInsert(c)
 	case "PATCH":
 		TestPatch(c)
+	case "REMOVE":
+		TestRemove(c)
 	default:
 		log.Fatalf("Unknown test %s", c.Test)
 	}

cmd/bench/main_test.go

@@ -0,0 +1,21 @@
+package main
+
+import (
+	"testing"
+)
+
+func Test_main(t *testing.T) {
+
+	t.Skip()
+
+	c := Config{
+		Test:    "remove",
+		Base:    "",
+		N:       1_000_000,
+		Workers: 4,
+	}
+
+	//TestRemove(c)
+	TestInsert(c)
+
+}

cmd/bench/test_remove.go

@@ -0,0 +1,148 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/fulldump/inceptiondb/bootstrap"
+	"github.com/fulldump/inceptiondb/collection"
+	"github.com/fulldump/inceptiondb/configuration"
+)
+
+func TestRemove(c Config) {
+
+	createServer := c.Base == ""
+
+	var start, stop func()
+	var dataDir string
+	if createServer {
+		dir, cleanup := TempDir()
+		dataDir = dir
+		cleanups = append(cleanups, cleanup)
+
+		conf := configuration.Default()
+		conf.Dir = dir
+		c.Base = "http://" + conf.HttpAddr
+
+		start, stop = bootstrap.Bootstrap(conf)
+		go start()
+	}
+
+	collectionName := CreateCollection(c.Base)
+
+	transport := &http.Transport{
+		MaxConnsPerHost:     1024,
+		MaxIdleConns:        1024,
+		MaxIdleConnsPerHost: 1024,
+	}
+	defer transport.CloseIdleConnections()
+
+	client := &http.Client{
+		Transport: transport,
+		Timeout:   10 * time.Second,
+	}
+
+	{
+		fmt.Println("Preload documents...")
+		r, w := io.Pipe()
+
+		encoder := json.NewEncoder(w)
+		go func() {
+			for i := int64(0); i < c.N; i++ {
+				encoder.Encode(JSON{
+					"id":     strconv.FormatInt(i, 10),
+					"value":  0,
+					"worker": i % int64(c.Workers),
+				})
+			}
+			w.Close()
+		}()
+
+		req, err := http.NewRequest("POST", c.Base+"/v1/collections/"+collectionName+":insert", r)
+		if err != nil {
+			fmt.Println("ERROR: new request:", err.Error())
+			os.Exit(3)
+		}
+
+		resp, err := client.Do(req)
+		if err != nil {
+			fmt.Println("ERROR: do request:", err.Error())
+			os.Exit(4)
+		}
+		io.Copy(io.Discard, resp.Body)
+	}
+
+	req, err := http.NewRequest("POST", c.Base+"/v1/collections/"+collectionName+":createIndex", strings.NewReader(`
+		{
+			"fields": ["id"],
+			"name": "my-index",
+			"type": "btree"
+		}`))
+	if err != nil {
+		fmt.Println("ERROR: new request:", err.Error())
+		os.Exit(3)
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		fmt.Println("ERROR: do request:", err.Error())
+		os.Exit(4)
+	}
+	io.Copy(io.Discard, resp.Body)
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusCreated {
+		fmt.Println("ERROR: bad status:", resp.Status)
+		os.Exit(5)
+	}
+
+	removeURL := fmt.Sprintf("%s/v1/collections/%s:remove", c.Base, collectionName)
+
+	t0 := time.Now()
+	worker := int64(-1)
+	Parallel(c.Workers, func() {
+		w := atomic.AddInt64(&worker, 1)
+
+		body := fmt.Sprintf(`{"filter":{"worker":%d},"limit":-1}`, w)
+		req, err := http.NewRequest(http.MethodPost, removeURL, strings.NewReader(body))
+		if err != nil {
+			fmt.Println("ERROR: new request:", err.Error())
+		}
+		req.Header.Set("Content-Type", "application/json")
+
+		resp, err := client.Do(req)
+		if err != nil {
+			fmt.Println("ERROR: do request:", err.Error())
+		}
+		io.Copy(io.Discard, resp.Body)
+		resp.Body.Close()
+
+		if resp.StatusCode != http.StatusOK {
+			fmt.Println("ERROR: bad status:", resp.Status)
+		}
+	})
+
+	took := time.Since(t0)
+	fmt.Println("sent:", c.N)
+	fmt.Println("took:", took)
+	fmt.Printf("Throughput: %.2f rows/sec\n", float64(c.N)/took.Seconds())
+
+	if !createServer {
+		return
+	}
+
+	stop() // Stop the server
+
+	t1 := time.Now()
+	collection.OpenCollection(path.Join(dataDir, collectionName))
+	tookOpen := time.Since(t1)
+	fmt.Println("open took:", tookOpen)
+	fmt.Printf("Throughput Open: %.2f rows/sec\n", float64(c.N)/tookOpen.Seconds())
+}