fix: synchronize notification user configuration cache access

notification-dingtalk/dingtalk_notification.go

@@ -63,12 +63,7 @@ func (n *Notification) Info() plugin.Info {
 
 // GetNewQuestionSubscribers returns the subscribers of the new question notification
 func (n *Notification) GetNewQuestionSubscribers() (userIDs []string) {
-	for userID, conf := range n.UserConfigCache.userConfigMapping {
-		if conf.AllNewQuestions {
-			userIDs = append(userIDs, userID)
-		}
-	}
-	return userIDs
+	return n.UserConfigCache.NewQuestionSubscriberIDs()
 }
 
 // Notify sends a notification to the user

notification-dingtalk/user_config.go

@@ -39,7 +39,7 @@ type UserConfig struct {
 type UserConfigCache struct {
 	// key: userID value: user config
 	userConfigMapping map[string]*UserConfig
-	sync.Mutex
+	sync.RWMutex
 }
 
 func NewUserConfigCache() *UserConfigCache {
@@ -55,6 +55,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) {
 	ucc.userConfigMapping[userID] = config
 }
 
+func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string {
+	ucc.RLock()
+	defer ucc.RUnlock()
+
+	userIDs := make([]string, 0)
+	for userID, conf := range ucc.userConfigMapping {
+		if conf.AllNewQuestions {
+			userIDs = append(userIDs, userID)
+		}
+	}
+	return userIDs
+}
+
 func (n *Notification) UserConfigFields() []plugin.ConfigField {
 	fields := make([]plugin.ConfigField, 0)
 	// Show tip for user, if the notification service is disabled

notification-dingtalk/user_config_concurrency_test.go

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package dingtalk
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+)
+
+func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) {
+	cache := NewUserConfigCache()
+	const workers, iterations = 8, 500
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	for worker := range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for i := range iterations {
+				cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0})
+			}
+		}()
+	}
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for range iterations {
+				_ = cache.NewQuestionSubscriberIDs()
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true})
+	for _, userID := range cache.NewQuestionSubscriberIDs() {
+		if userID == "subscriber" {
+			return
+		}
+	}
+	t.Fatal("subscriber missing from snapshot")
+}

notification-lark/notification.go

@@ -211,12 +211,7 @@ func makeCardMsg(args plugin.NotificationMessage) Card {
 
 // GetNewQuestionSubscribers returns the subscribers of the new question notification
 func (n *Notification) GetNewQuestionSubscribers() (userIDs []string) {
-	for userID, conf := range n.userConfigCache.userConfigMapping {
-		if conf.AllNewQuestions {
-			userIDs = append(userIDs, userID)
-		}
-	}
-	return userIDs
+	return n.userConfigCache.NewQuestionSubscriberIDs()
 }
 
 // Notify sends a notification to the user

notification-lark/user_config.go

@@ -39,7 +39,7 @@ type UserConfig struct {
 
 type UserConfigCache struct {
 	userConfigMapping map[string]*UserConfig
-	sync.Mutex
+	sync.RWMutex
 }
 
 func NewUserConfigCache() *UserConfigCache {
@@ -55,6 +55,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) {
 	ucc.userConfigMapping[userID] = config
 }
 
+func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string {
+	ucc.RLock()
+	defer ucc.RUnlock()
+
+	userIDs := make([]string, 0)
+	for userID, conf := range ucc.userConfigMapping {
+		if conf.AllNewQuestions {
+			userIDs = append(userIDs, userID)
+		}
+	}
+	return userIDs
+}
+
 func (n *Notification) UserConfigFields() []plugin.ConfigField {
 	return []plugin.ConfigField{
 		{

notification-lark/user_config_concurrency_test.go

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package lark
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+)
+
+func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) {
+	cache := NewUserConfigCache()
+	const workers, iterations = 8, 500
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	for worker := range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for i := range iterations {
+				cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0})
+			}
+		}()
+	}
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for range iterations {
+				_ = cache.NewQuestionSubscriberIDs()
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true})
+	for _, userID := range cache.NewQuestionSubscriberIDs() {
+		if userID == "subscriber" {
+			return
+		}
+	}
+	t.Fatal("subscriber missing from snapshot")
+}

notification-slack/slack_notification.go

@@ -63,12 +63,7 @@ func (n *Notification) Info() plugin.Info {
 
 // GetNewQuestionSubscribers returns the subscribers of the new question notification
 func (n *Notification) GetNewQuestionSubscribers() (userIDs []string) {
-	for userID, conf := range n.UserConfigCache.userConfigMapping {
-		if conf.AllNewQuestions {
-			userIDs = append(userIDs, userID)
-		}
-	}
-	return userIDs
+	return n.UserConfigCache.NewQuestionSubscriberIDs()
 }
 
 // Notify sends a notification to the user

notification-slack/user_config.go

@@ -43,7 +43,7 @@ type UserConfig struct {
 type UserConfigCache struct {
 	// key: userID value: user config
 	userConfigMapping map[string]*UserConfig
-	sync.Mutex
+	sync.RWMutex
 }
 
 func NewUserConfigCache() *UserConfigCache {
@@ -59,6 +59,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) {
 	ucc.userConfigMapping[userID] = config
 }
 
+func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string {
+	ucc.RLock()
+	defer ucc.RUnlock()
+
+	userIDs := make([]string, 0)
+	for userID, conf := range ucc.userConfigMapping {
+		if conf.AllNewQuestions {
+			userIDs = append(userIDs, userID)
+		}
+	}
+	return userIDs
+}
+
 func (n *Notification) UserConfigFields() []plugin.ConfigField {
 	fields := make([]plugin.ConfigField, 0)
 	// Show tip for user, if the notification service is disabled

notification-slack/user_config_concurrency_test.go

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package slack_notification
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+)
+
+func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) {
+	cache := NewUserConfigCache()
+	const workers, iterations = 8, 500
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	for worker := range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for i := range iterations {
+				cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0})
+			}
+		}()
+	}
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for range iterations {
+				_ = cache.NewQuestionSubscriberIDs()
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true})
+	for _, userID := range cache.NewQuestionSubscriberIDs() {
+		if userID == "subscriber" {
+			return
+		}
+	}
+	t.Fatal("subscriber missing from snapshot")
+}

notification-wecom/notification_wecom.go

@@ -62,12 +62,7 @@ func (*Notification) Info() plugin.Info {
 
 // GetNewQuestionSubscribers returns the subscribers of the new question notification
 func (n *Notification) GetNewQuestionSubscribers() (userIDs []string) {
-	for userID, conf := range n.UserConfigCache.userConfigMapping {
-		if conf.AllNewQuestions {
-			userIDs = append(userIDs, userID)
-		}
-	}
-	return userIDs
+	return n.UserConfigCache.NewQuestionSubscriberIDs()
 }
 
 // Notify sends a notification to the user

notification-wecom/user_config.go

@@ -39,7 +39,7 @@ type UserConfig struct {
 type UserConfigCache struct {
 	// key: userID value: user config
 	userConfigMapping map[string]*UserConfig
-	sync.Mutex
+	sync.RWMutex
 }
 
 func NewUserConfigCache() *UserConfigCache {
@@ -55,6 +55,19 @@ func (ucc *UserConfigCache) SetUserConfig(userID string, config *UserConfig) {
 	ucc.userConfigMapping[userID] = config
 }
 
+func (ucc *UserConfigCache) NewQuestionSubscriberIDs() []string {
+	ucc.RLock()
+	defer ucc.RUnlock()
+
+	userIDs := make([]string, 0)
+	for userID, conf := range ucc.userConfigMapping {
+		if conf.AllNewQuestions {
+			userIDs = append(userIDs, userID)
+		}
+	}
+	return userIDs
+}
+
 func (n *Notification) UserConfigFields() []plugin.ConfigField {
 	fields := make([]plugin.ConfigField, 0)
 	// Show tip for user, if the notification service is disabled

notification-wecom/user_config_concurrency_test.go

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package wecom
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+)
+
+func TestUserConfigCacheConcurrentNewQuestionSubscribers(t *testing.T) {
+	cache := NewUserConfigCache()
+	const workers, iterations = 8, 500
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	for worker := range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for i := range iterations {
+				cache.SetUserConfig(fmt.Sprintf("writer-%d-user-%d", worker, i%32), &UserConfig{AllNewQuestions: i%2 == 0})
+			}
+		}()
+	}
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for range iterations {
+				_ = cache.NewQuestionSubscriberIDs()
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	cache.SetUserConfig("subscriber", &UserConfig{AllNewQuestions: true})
+	for _, userID := range cache.NewQuestionSubscriberIDs() {
+		if userID == "subscriber" {
+			return
+		}
+	}
+	t.Fatal("subscriber missing from snapshot")
+}

user-center-slack/notification.go

@@ -31,12 +31,7 @@ import (
 
 // GetNewQuestionSubscribers returns the subscribers of the new question notification
 func (uc *UserCenter) GetNewQuestionSubscribers() (userIDs []string) {
-	for userID, conf := range uc.UserConfigCache.userConfigMapping {
-		if conf.AllNewQuestions {
-			userIDs = append(userIDs, userID)
-		}
-	}
-	return userIDs
+	return uc.UserConfigCache.NewQuestionSubscriberIDs()
 }
 
 // Notify sends a notification to the user