Introduce KV mutex

Author: k-yomo

GNUmakefile

@@ -14,7 +14,7 @@ generate:
 # Run acceptance tests
 .PHONY: testacc
 testacc:
-	TF_ACC=1 go test ./... -v $(TESTARGS) -coverprofile=coverage.out -timeout 120m
+	TF_ACC=1 go test ./... -v $(TESTARGS) -race -coverprofile=coverage.out -timeout 120m
 
 # Run acceptance tests and show coverages
 .PHONY: testacc-cover

internal/mutex/kv.go

@@ -0,0 +1,49 @@
+package mutex
+
+import (
+	"log"
+	"sync"
+)
+
+// KV is a simple key/value store for arbitrary mutexes. It can be used to
+// serialize changes across arbitrary collaborators that share knowledge of the
+// keys they must serialize on.
+//
+type KV struct {
+	lock  sync.Mutex
+	store map[string]*sync.Mutex
+}
+
+// NewKV returns a properly initialized KV
+func NewKV() *KV {
+	return &KV{
+		store: make(map[string]*sync.Mutex),
+	}
+}
+
+// Lock the mutex for the given key. Caller is responsible for calling Unlock
+// for the same key
+func (m *KV) Lock(key string) {
+	log.Printf("[DEBUG] Locking %q", key)
+	m.get(key).Lock()
+	log.Printf("[DEBUG] Locked %q", key)
+}
+
+// Unlock the mutex for the given key. Caller must have called Lock for the same key first
+func (m *KV) Unlock(key string) {
+	log.Printf("[DEBUG] Unlocking %q", key)
+	m.get(key).Unlock()
+	log.Printf("[DEBUG] Unlocked %q", key)
+}
+
+// Returns a mutex for the given key, no guarantee of its lock status
+func (m *KV) get(key string) *sync.Mutex {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	mutex, ok := m.store[key]
+	if !ok {
+		mutex = &sync.Mutex{}
+		m.store[key] = mutex
+	}
+	return mutex
+}

internal/mutex/kv_test.go

@@ -0,0 +1,38 @@
+package mutex
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestKV(t *testing.T) {
+	t.Parallel()
+
+	mutexKV := NewKV()
+	start := time.Now()
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		mutexKV.Lock("test")
+		time.Sleep(100 * time.Millisecond)
+		mutexKV.Unlock("test")
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		mutexKV.Lock("test")
+		time.Sleep(100 * time.Millisecond)
+		mutexKV.Unlock("test")
+	}()
+	wg.Wait()
+
+	if elapsed := time.Since(start); elapsed < 200*time.Millisecond {
+		t.Errorf("TestKV() elapsed time = %v, want %v", elapsed, 200*time.Millisecond)
+	}
+}

internal/provider/provider.go

@@ -2,7 +2,6 @@ package provider
 
 import (
 	"context"
-
 	"github.com/algolia/algoliasearch-client-go/v3/algolia/region"
 	"github.com/algolia/algoliasearch-client-go/v3/algolia/search"
 	"github.com/algolia/algoliasearch-client-go/v3/algolia/suggestions"
@@ -11,8 +10,12 @@ import (
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/logging"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-provider-algolia/internal/algoliautil"
+	"github.com/hashicorp/terraform-provider-algolia/internal/mutex"
 )
 
+// Global Key/Value Mutex
+var mutexKV = mutex.NewKV()
+
 // nolint: gochecknoinits
 func init() {
 	schema.DescriptionKind = schema.StringMarkdown

internal/provider/resource_index.go

@@ -2,6 +2,7 @@ package provider
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"strconv"
 	"time"
@@ -583,8 +584,14 @@ func resourceIndexCreate(ctx context.Context, d *schema.ResourceData, m interfac
 
 	indexName := d.Get("name").(string)
 
-	if primaryIndexName, ok := d.GetOk("primary_index_name"); ok {
-		primaryIndex := apiClient.searchClient.InitIndex(primaryIndexName.(string))
+	if v, ok := d.GetOk("primary_index_name"); ok {
+		primaryIndexName := v.(string)
+		// Modifying the primary's replica setting on primary can cause problems if other replicas
+		// are modifying it at the same time. Lock the primary until we're done in order to prevent that.
+		mutexKV.Lock(algoliaIndexMutexKey(apiClient.appID, primaryIndexName))
+		defer mutexKV.Unlock(algoliaIndexMutexKey(apiClient.appID, primaryIndexName))
+
+		primaryIndex := apiClient.searchClient.InitIndex(primaryIndexName)
 		primaryIndexSettings, err := primaryIndex.GetSettings(ctx)
 		if err != nil {
 			return diag.FromErr(err)
@@ -653,8 +660,14 @@ func resourceIndexDelete(ctx context.Context, d *schema.ResourceData, m interfac
 	apiClient := m.(*apiClient)
 	indexName := d.Id()
 
-	if primaryIndexName, ok := d.GetOk("primary_index_name"); ok {
-		primaryIndex := apiClient.searchClient.InitIndex(primaryIndexName.(string))
+	if v, ok := d.GetOk("primary_index_name"); ok {
+		primaryIndexName := v.(string)
+		// Modifying the primary's replica setting on primary can cause problems if other replicas
+		// are modifying it at the same time. Lock the primary until we're done in order to prevent that.
+		mutexKV.Lock(algoliaIndexMutexKey(apiClient.appID, primaryIndexName))
+		defer mutexKV.Unlock(algoliaIndexMutexKey(apiClient.appID, primaryIndexName))
+
+		primaryIndex := apiClient.searchClient.InitIndex(primaryIndexName)
 		primaryIndexSettings, err := primaryIndex.GetSettings(ctx)
 		if err != nil {
 			return diag.FromErr(err)
@@ -1224,3 +1237,7 @@ func unmarshalAdvancedConfig(configured interface{}, settings *search.Settings,
 		}
 	}
 }
+
+func algoliaIndexMutexKey(appID string, indexName string) string {
+	return fmt.Sprintf("%s-algolia-index-%s", appID, indexName)
+}

internal/provider/resource_virtual_index.go

@@ -583,6 +583,11 @@ func resourceVirtualIndexCreate(ctx context.Context, d *schema.ResourceData, m i
 		}
 	}
 	if !virtualReplicaExist {
+		// Modifying the primary's replica setting on primary can cause problems if other replicas
+		// are modifying it at the same time. Lock the primary until we're done in order to prevent that.
+		mutexKV.Lock(algoliaIndexMutexKey(apiClient.appID, primaryIndexName))
+		defer mutexKV.Unlock(algoliaIndexMutexKey(apiClient.appID, primaryIndexName))
+
 		newReplicas := append(primaryIndexSettings.Replicas.Get(), fmt.Sprintf("virtual(%s)", indexName))
 		res, err := primaryIndex.SetSettings(search.Settings{
 			Replicas: opt.Replicas(newReplicas...),
@@ -640,6 +645,11 @@ func resourceVirtualIndexDelete(ctx context.Context, d *schema.ResourceData, m i
 	indexName := d.Id()
 
 	primaryIndexName := d.Get("primary_index_name").(string)
+	// Modifying the primary's replica setting on primary can cause problems if other replicas
+	// are modifying it at the same time. Lock the primary until we're done in order to prevent that.
+	mutexKV.Lock(algoliaIndexMutexKey(apiClient.appID, primaryIndexName))
+	defer mutexKV.Unlock(algoliaIndexMutexKey(apiClient.appID, primaryIndexName))
+
 	primaryIndex := apiClient.searchClient.InitIndex(primaryIndexName)
 	primaryIndexSettings, err := primaryIndex.GetSettings(ctx)
 	if err != nil {