mutex lock namespace and issues to prevent concurrent hook process

Author: bivas

bot/bot.go

@@ -3,9 +3,12 @@ package bot
 import (
 	"fmt"
 	"github.com/bivas/rivi/util"
+	"github.com/patrickmn/go-cache"
 	"net/http"
 	"path/filepath"
 	"strings"
+	"sync"
+	"time"
 )
 
 type HandledEventResult struct {
@@ -20,6 +23,10 @@ type Bot interface {
 type bot struct {
 	defaultNamespace string
 	configurations   map[string]Configuration
+
+	cacheLocker      *sync.Mutex
+	namespaceMutexes *cache.Cache
+	repoIssueMutexes *cache.Cache
 }
 
 func (b *bot) getCurrentConfiguration(namespace string) (Configuration, error) {
@@ -34,21 +41,25 @@ func (b *bot) getCurrentConfiguration(namespace string) (Configuration, error) {
 	return configuration, nil
 }
 
-func (b *bot) HandleEvent(r *http.Request) *HandledEventResult {
-
-	workingConfiguration, err := b.getCurrentConfiguration(r.URL.Query().Get("namespace"))
-	if err != nil {
-		return &HandledEventResult{Message: err.Error()}
-	}
-	data, process := buildFromRequest(workingConfiguration.GetClientConfig(), r)
-	if !process {
-		return &HandledEventResult{Message: "Skipping rules processing (could be not supported event type)"}
+func (b *bot) processRules(configuration Configuration, data EventData) *HandledEventResult {
+	id := fmt.Sprintf("%s/%s#%d", data.GetOwner(), data.GetRepo(), data.GetNumber())
+	util.Logger.Debug("acquire global lock during rules process")
+	b.cacheLocker.Lock()
+	locker, exists := b.repoIssueMutexes.Get(id)
+	if !exists {
+		locker = &sync.Mutex{}
+		b.repoIssueMutexes.Set(id, locker, cache.DefaultExpiration)
 	}
+	util.Logger.Debug("acquire repo issue %s lock during rules process", id)
+	locker.(*sync.Mutex).Lock()
+	defer locker.(*sync.Mutex).Unlock()
+	util.Logger.Debug("release global lock during rules process")
+	b.cacheLocker.Unlock()
 	applied := make([]Rule, 0)
 	result := &HandledEventResult{
 		AppliedRules: []string{},
 	}
-	for _, rule := range workingConfiguration.GetRules() {
+	for _, rule := range configuration.GetRules() {
 		if rule.Accept(data) {
 			util.Logger.Debug("Accepting rule %s for '%s'", rule.Name(), data.GetTitle())
 			applied = append(applied, rule)
@@ -58,14 +69,45 @@ func (b *bot) HandleEvent(r *http.Request) *HandledEventResult {
 	for _, rule := range applied {
 		util.Logger.Debug("Applying rule %s for '%s'", rule.Name(), data.GetTitle())
 		for _, action := range rule.Actions() {
-			action.Apply(workingConfiguration, data)
+			action.Apply(configuration, data)
 		}
 	}
 	return result
 }
 
+func (b *bot) HandleEvent(r *http.Request) *HandledEventResult {
+	namespace := r.URL.Query().Get("namespace")
+	b.cacheLocker.Lock()
+	util.Logger.Debug("acquire global lock during namespace process")
+	locker, exists := b.namespaceMutexes.Get(namespace)
+	if !exists {
+		locker = &sync.Mutex{}
+		b.namespaceMutexes.Set(namespace, locker, cache.DefaultExpiration)
+	}
+	util.Logger.Debug("acquire namespace %s lock", namespace)
+	locker.(*sync.Mutex).Lock()
+	util.Logger.Debug("release global lock during namespace process")
+	b.cacheLocker.Unlock()
+	workingConfiguration, err := b.getCurrentConfiguration(namespace)
+	if err != nil {
+		return &HandledEventResult{Message: err.Error()}
+	}
+	data, process := buildFromRequest(workingConfiguration.GetClientConfig(), r)
+	if !process {
+		return &HandledEventResult{Message: "Skipping rules processing (could be not supported event type)"}
+	}
+	util.Logger.Debug("release namespace %s lock", namespace)
+	locker.(*sync.Mutex).Unlock()
+	return b.processRules(workingConfiguration, data)
+}
+
 func New(configPaths ...string) (Bot, error) {
-	b := &bot{configurations: make(map[string]Configuration)}
+	b := &bot{
+		configurations:   make(map[string]Configuration),
+		cacheLocker:      &sync.Mutex{},
+		namespaceMutexes: cache.New(time.Minute, 30*time.Second),
+		repoIssueMutexes: cache.New(time.Minute, 20*time.Second),
+	}
 	for index, configPath := range configPaths {
 		baseConfigPath := filepath.Base(configPath)
 		namespace := strings.TrimSuffix(baseConfigPath, filepath.Ext(baseConfigPath))