Function TestConcurrency

plugin/plugin_test/kv_storage_test.go:301–412

Test concurrent operations and performance

(t *testing.T)

// Test concurrent operations and performance
func TestConcurrency(t *testing.T) {
	setupTestEnvironment()
	kv := testPlugin.operator
	ctx := context.Background()

	t.Run("BasicConcurrency", func(t *testing.T) {
		parallel := 10
		var wg sync.WaitGroup
		wg.Add(parallel)

		for i := range parallel {
			go func(index int) {
				defer wg.Done()
				time.Sleep(time.Duration(rand.Intn(200)) * time.Millisecond)
				mustSet(t, kv, ctx, "concurrent", fmt.Sprintf("key%d", index), "value")
			}(i)
		}
		wg.Wait()

		// Verify results
		wg.Add(parallel)
		for i := range parallel {
			go func(index int) {
				defer wg.Done()
				time.Sleep(time.Duration(rand.Intn(200)) * time.Millisecond)
				mustGet(t, kv, ctx, "concurrent", fmt.Sprintf("key%d", index), "value")
			}(i)
		}
		wg.Wait()
	})

	t.Run("StressTest", func(t *testing.T) {
		if testing.Short() {
			t.Skip("Skipping stress test in short mode")
		}

		totalOps := 1000
		workerCount := 20
		prefix := "stress_test"
		opsPerWorker := totalOps / workerCount

		log.Info("Starting KV storage stress test...")
		startTime := time.Now()

		// Concurrent write test
		var wg sync.WaitGroup
		errorCount := int64(0)

		for w := range workerCount {
			wg.Add(1)
			go func(workerID int) {
				defer wg.Done()
				startIdx := workerID * opsPerWorker

				for i := range opsPerWorker {
					i := startIdx + i
					err := kv.Set(ctx, plugin.KVParams{
						Group: prefix,