Sync plugin settings and status across replicas

internal/controller_admin/plugin_controller.go

@@ -20,8 +20,6 @@
 package controller_admin
 
 import (
-	"encoding/json"
-
 	"github.com/apache/answer/internal/base/handler"
 	"github.com/apache/answer/internal/schema"
 	"github.com/apache/answer/internal/service/plugin_common"
@@ -148,8 +146,7 @@ func (pc *PluginController) UpdatePluginStatus(ctx *gin.Context) {
 		return
 	}
 
-	plugin.StatusManager.Enable(req.PluginSlugName, req.Enabled)
-	err := pc.pluginCommonService.UpdatePluginStatus(ctx)
+	err := pc.pluginCommonService.UpdatePluginStatus(ctx, req.PluginSlugName, req.Enabled)
 	handler.HandleResponse(ctx, err, nil)
 }
 
@@ -207,18 +204,6 @@ func (pc *PluginController) UpdatePluginConfig(ctx *gin.Context) {
 		return
 	}
 
-	configFields, _ := json.Marshal(req.ConfigFields)
-	err := plugin.CallConfig(func(fn plugin.Config) error {
-		if fn.Info().SlugName == req.PluginSlugName {
-			return fn.ConfigReceiver(configFields)
-		}
-		return nil
-	})
-	if err != nil {
-		handler.HandleResponse(ctx, err, nil)
-		return
-	}
-
-	err = pc.pluginCommonService.UpdatePluginConfig(ctx, req)
+	err := pc.pluginCommonService.UpdatePluginConfig(ctx, req)
 	handler.HandleResponse(ctx, err, nil)
 }

internal/repo/config/config_repo.go

@@ -141,3 +141,44 @@ func (cr configRepo) UpdateConfig(ctx context.Context, key string, value string)
 	}
 	return
 }
+
+func (cr configRepo) CompareAndSwapConfig(
+	ctx context.Context, key, oldValue, newValue string,
+) (swapped bool, err error) {
+	oldConfig := &entity.Config{Key: key}
+	exist, err := cr.data.DB.Context(ctx).Get(oldConfig)
+	if err != nil {
+		return false, errors.InternalServer(reason.DatabaseError).WithError(err).WithStack()
+	}
+	if !exist {
+		return false, errors.BadRequest(reason.ObjectNotFound)
+	}
+	if oldConfig.Value != oldValue {
+		return false, nil
+	}
+
+	affected, err := cr.data.DB.Context(ctx).ID(oldConfig.ID).Where("value = ?", oldValue).
+		Update(&entity.Config{Value: newValue})
+	if err != nil {
+		return false, errors.InternalServer(reason.DatabaseError).WithError(err).WithStack()
+	}
+	if affected == 0 {
+		if oldValue != newValue {
+			return false, nil
+		}
+		current, err := cr.GetConfigByKeyFromDB(ctx, key)
+		if err != nil {
+			return false, err
+		}
+		return current.Value == newValue, nil
+	}
+
+	// Invalidate both cached forms. The next reader will load the latest DB value.
+	if err := cr.data.Cache.Del(ctx, constant.ConfigKEY2ContentCacheKeyPrefix+key); err != nil {
+		log.Error(err)
+	}
+	if err := cr.data.Cache.Del(ctx, fmt.Sprintf("%s%d", constant.ConfigID2KEYCacheKeyPrefix, oldConfig.ID)); err != nil {
+		log.Error(err)
+	}
+	return true, nil
+}

internal/service/config/config_service.go

@@ -33,6 +33,7 @@ type ConfigRepo interface {
 	GetConfigByKey(ctx context.Context, key string) (c *entity.Config, err error)
 	GetConfigByKeyFromDB(ctx context.Context, key string) (c *entity.Config, err error)
 	UpdateConfig(ctx context.Context, key, value string) (err error)
+	CompareAndSwapConfig(ctx context.Context, key, oldValue, newValue string) (swapped bool, err error)
 }
 
 // ConfigService user service
@@ -116,3 +117,11 @@ func (cs *ConfigService) GetIDByKey(ctx context.Context, key string) (id int, er
 func (cs *ConfigService) UpdateConfig(ctx context.Context, key, value string) (err error) {
 	return cs.configRepo.UpdateConfig(ctx, key, value)
 }
+
+// CompareAndSwapConfig updates a config only when its current value still
+// matches oldValue.
+func (cs *ConfigService) CompareAndSwapConfig(
+	ctx context.Context, key, oldValue, newValue string,
+) (swapped bool, err error) {
+	return cs.configRepo.CompareAndSwapConfig(ctx, key, oldValue, newValue)
+}

internal/service/notification/new_question_notification_test.go

@@ -674,6 +674,12 @@ func (newQuestionNotificationTestConfigRepo) UpdateConfig(context.Context, strin
 	return nil
 }
 
+func (newQuestionNotificationTestConfigRepo) CompareAndSwapConfig(
+	context.Context, string, string, string,
+) (bool, error) {
+	return false, nil
+}
+
 type newQuestionNotificationTestEmailRepo struct {
 	codesByUserID map[string][]string
 }

internal/service/plugin_common/plugin_common_service.go

@@ -22,6 +22,9 @@ package plugin_common
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"sync"
+	"time"
 
 	"github.com/apache/answer/internal/base/data"
 	"github.com/apache/answer/internal/repo/search_sync"
@@ -60,8 +63,16 @@ type PluginCommonService struct {
 	pluginUserConfigRepo PluginUserConfigRepo
 	data                 *data.Data
 	importerService      *importer.ImporterService
+	pluginStatusMu       sync.Mutex
+	pluginConfigMu       sync.Mutex
+	pluginConfigValues   map[string]string
 }
 
+const (
+	pluginConfigRefreshInterval = 10 * time.Second
+	pluginStatusUpdateAttempts  = 10
+)
+
 // NewPluginCommonService new report service
 func NewPluginCommonService(
 	pluginConfigRepo PluginConfigRepo,
@@ -76,36 +87,91 @@ func NewPluginCommonService(
 		pluginUserConfigRepo: pluginUserConfigRepo,
 		data:                 data,
 		importerService:      importerService,
+		pluginConfigValues:   make(map[string]string),
 	}
 	p.initPluginData()
+	p.startPluginConfigRefresh()
 	return p
 }
 
 // UpdatePluginStatus update plugin status
-func (ps *PluginCommonService) UpdatePluginStatus(ctx context.Context) (err error) {
-	content, err := plugin.StatusManager.MarshalJSON()
-	if err != nil {
-		return errors.InternalServer(reason.UnknownError).WithError(err)
+func (ps *PluginCommonService) UpdatePluginStatus(ctx context.Context, pluginSlugName string, enabled bool) error {
+	ps.pluginStatusMu.Lock()
+	defer ps.pluginStatusMu.Unlock()
+
+	for range pluginStatusUpdateAttempts {
+		current, err := ps.configService.GetStringValueFromDB(ctx, constant.PluginStatus)
+		if err != nil {
+			return err
+		}
+
+		updated, err := plugin.StatusManager.UpdateJSON([]byte(current), pluginSlugName, enabled)
+		if err != nil {
+			return errors.InternalServer(reason.UnknownError).WithError(err)
+		}
+
+		swapped, err := ps.configService.CompareAndSwapConfig(
+			ctx, constant.PluginStatus, current, string(updated))
+		if err != nil {
+			return err
+		}
+		if !swapped {
+			continue
+		}
+
+		if err := plugin.StatusManager.UnmarshalJSON(updated); err != nil {
+			return errors.InternalServer(reason.UnknownError).WithError(err)
+		}
+		return nil
 	}
-	return ps.configService.UpdateConfig(ctx, constant.PluginStatus, string(content))
+
+	return errors.InternalServer(reason.UnknownError).WithError(
+		fmt.Errorf("plugin status changed concurrently; retry the request"))
 }
 
 // UpdatePluginConfig update plugin config
 func (ps *PluginCommonService) UpdatePluginConfig(ctx context.Context, req *schema.UpdatePluginConfigReq) (err error) {
-	configValue, _ := json.Marshal(req.ConfigFields)
+	configValue, err := json.Marshal(req.ConfigFields)
+	if err != nil {
+		return errors.InternalServer(reason.UnknownError).WithError(err)
+	}
+
+	ps.pluginConfigMu.Lock()
+	err = plugin.CallConfig(func(fn plugin.Config) error {
+		if fn.Info().SlugName == req.PluginSlugName {
+			return fn.ConfigReceiver(configValue)
+		}
+		return nil
+	})
+	if err != nil {
+		ps.pluginConfigMu.Unlock()
+		return err
+	}
+
 	err = ps.pluginConfigRepo.SavePluginConfig(ctx, req.PluginSlugName, string(configValue))
 	if err != nil {
+		ps.pluginConfigMu.Unlock()
 		return err
 	}
+	if ps.pluginConfigValues == nil {
+		ps.pluginConfigValues = make(map[string]string)
+	}
+	ps.pluginConfigValues[req.PluginSlugName] = string(configValue)
+	ps.pluginConfigMu.Unlock()
 
+	ps.registerPluginConfigDependents(ctx, req.PluginSlugName)
+	return nil
+}
+
+func (ps *PluginCommonService) registerPluginConfigDependents(ctx context.Context, pluginSlugName string) {
 	_ = plugin.CallSearch(func(search plugin.Search) error {
-		if search.Info().SlugName == req.PluginSlugName {
+		if search.Info().SlugName == pluginSlugName {
 			search.RegisterSyncer(ctx, search_sync.NewPluginSyncer(ps.data))
 		}
 		return nil
 	})
 	_ = plugin.CallVectorSearch(func(vs plugin.VectorSearch) error {
-		if vs.Info().SlugName == req.PluginSlugName {
+		if vs.Info().SlugName == pluginSlugName {
 			vs.RegisterSyncer(ctx, vector_search_sync.NewPluginSyncer(ps.data))
 		}
 		return nil
@@ -114,7 +180,6 @@ func (ps *PluginCommonService) UpdatePluginConfig(ctx context.Context, req *sche
 		importer.RegisterImporterFunc(ctx, ps.importerService.NewImporterFunc())
 		return nil
 	})
-	return nil
 }
 
 // UpdatePluginUserConfig update plugin config
@@ -150,38 +215,14 @@ func (ps *PluginCommonService) initPluginData() {
 		return nil
 	})
 
-	// init plugin status
-	pluginStatus, err := ps.configService.GetStringValueFromDB(context.TODO(), constant.PluginStatus)
-	if err != nil {
-		log.Error(err)
-	} else {
-		if err := plugin.StatusManager.UnmarshalJSON([]byte(pluginStatus)); err != nil {
-			log.Error(err)
-		}
-	}
-
-	// init plugin config
-	pluginConfigs, err := ps.pluginConfigRepo.GetPluginConfigAll(context.Background())
-	if err != nil {
+	ps.refreshPluginStatus(context.Background())
+	if err := ps.refreshPluginConfigs(context.Background(), true, false); err != nil {
 		log.Error(err)
-	} else {
-		for _, pluginConfig := range pluginConfigs {
-			err := plugin.CallConfig(func(fn plugin.Config) error {
-				if fn.Info().SlugName == pluginConfig.PluginSlugName {
-					return fn.ConfigReceiver([]byte(pluginConfig.Value))
-				}
-				return nil
-			})
-			if err != nil {
-				log.Errorf("parse plugin config failed: %s %v", pluginConfig.PluginSlugName, err)
-			}
-		}
-
-		_ = plugin.CallCache(func(cache plugin.Cache) error {
-			ps.data.Cache = cache
-			return nil
-		})
 	}
+	_ = plugin.CallCache(func(cache plugin.Cache) error {
+		ps.data.Cache = cache
+		return nil
+	})
 
 	// register syncer for vector search plugins on startup
 	_ = plugin.CallVectorSearch(func(vs plugin.VectorSearch) error {
@@ -229,3 +270,73 @@ func (ps *PluginCommonService) initPluginData() {
 		}
 	}()
 }
+
+func (ps *PluginCommonService) startPluginConfigRefresh() {
+	ticker := time.NewTicker(pluginConfigRefreshInterval)
+	go func() {
+		defer ticker.Stop()
+		for range ticker.C {
+			ps.refreshPluginStatus(context.Background())
+			if err := ps.refreshPluginConfigs(context.Background(), false, true); err != nil {
+				log.Error(err)
+			}
+		}
+	}()
+}
+
+func (ps *PluginCommonService) refreshPluginStatus(ctx context.Context) {
+	ps.pluginStatusMu.Lock()
+	defer ps.pluginStatusMu.Unlock()
+
+	pluginStatus, err := ps.configService.GetStringValueFromDB(ctx, constant.PluginStatus)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+	if err := plugin.StatusManager.UnmarshalJSON([]byte(pluginStatus)); err != nil {
+		log.Error(err)
+	}
+}
+
+func (ps *PluginCommonService) refreshPluginConfigs(ctx context.Context, force, registerDependents bool) error {
+	ps.pluginConfigMu.Lock()
+	pluginConfigs, err := ps.pluginConfigRepo.GetPluginConfigAll(ctx)
+	if err != nil {
+		ps.pluginConfigMu.Unlock()
+		return err
+	}
+	if ps.pluginConfigValues == nil {
+		ps.pluginConfigValues = make(map[string]string)
+	}
+
+	changedPlugins := make([]string, 0, len(pluginConfigs))
+	for _, pluginConfig := range pluginConfigs {
+		if !force && ps.pluginConfigValues[pluginConfig.PluginSlugName] == pluginConfig.Value {
+			continue
+		}
+
+		applied := false
+		err := plugin.CallConfig(func(fn plugin.Config) error {
+			if fn.Info().SlugName != pluginConfig.PluginSlugName {
+				return nil
+			}
+			applied = true
+			return fn.ConfigReceiver([]byte(pluginConfig.Value))
+		})
+		if err != nil {
+			log.Errorf("parse plugin config failed: %s %v", pluginConfig.PluginSlugName, err)
+			continue
+		}
+		if applied {
+			ps.pluginConfigValues[pluginConfig.PluginSlugName] = pluginConfig.Value
+			changedPlugins = append(changedPlugins, pluginConfig.PluginSlugName)
+		}
+	}
+	ps.pluginConfigMu.Unlock()
+	if registerDependents {
+		for _, pluginSlugName := range changedPlugins {
+			ps.registerPluginConfigDependents(ctx, pluginSlugName)
+		}
+	}
+	return nil
+}