feat: use context aware semaphore instead of mutex

Author: christophwitzko

internal/server/handler_batch.go

@@ -137,9 +137,12 @@ func (s *Server) batchGetPlugins(w http.ResponseWriter, r *http.Request) {
 	batchResponse.DownloadURL = s.config.GetPublicPluginCacheDownloadURL(archiveKey)
 
 	// allow only one batch archive process at a time
-	// TODO: this might be to conservative, we could allow multiple archives to be created at the same time
-	s.batchMu.Lock()
-	defer s.batchMu.Unlock()
+	err = s.batchArchiveSemaphore.Acquire(r.Context(), 1)
+	if err != nil {
+		s.writeJSONError(w, r, http.StatusTooManyRequests, err, "could not acquire semaphore")
+		return
+	}
+	defer s.batchArchiveSemaphore.Release(1)
 
 	headRes, err := s.storage.HeadObject(r.Context(), &s3.HeadObjectInput{
 		Bucket: s.config.GetBucket(),

internal/server/handlers.go

@@ -17,9 +17,14 @@ func (s *Server) listPlugins(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) updateAllPlugins(w http.ResponseWriter, r *http.Request) {
-	s.ghMutex.Lock()
-	defer s.ghMutex.Unlock()
-	s.log.Warn("updating all plugins")
+	err := s.ghSemaphore.Acquire(r.Context(), 1)
+	if err != nil {
+		s.writeJSONError(w, r, http.StatusTooManyRequests, err, "could not acquire semaphore")
+		return
+	}
+	defer s.ghSemaphore.Release(1)
+
+	s.log.Warn("updating all plugins...")
 	for _, p := range config.Plugins {
 		s.log.Infof("updating plugin %s", p.GetFullName())
 		err := p.Update(r.Context(), s.db, s.ghClient, "")
@@ -33,8 +38,13 @@ func (s *Server) updateAllPlugins(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) updatePlugin(w http.ResponseWriter, r *http.Request) {
-	s.ghMutex.Lock()
-	defer s.ghMutex.Unlock()
+	err := s.ghSemaphore.Acquire(r.Context(), 1)
+	if err != nil {
+		s.writeJSONError(w, r, http.StatusTooManyRequests, err, "could not acquire semaphore")
+		return
+	}
+	defer s.ghSemaphore.Release(1)
+
 	pluginVersion := chi.URLParam(r, "version")
 	pluginName := chi.URLParam(r, "plugin")
 	p := config.Plugins.Find(pluginName)

internal/server/server.go

@@ -3,7 +3,6 @@ package server
 import (
 	"fmt"
 	"net/http"
-	"sync"
 	"time"
 
 	"cloud.google.com/go/firestore"
@@ -15,18 +14,20 @@ import (
 	"github.com/google/go-github/v50/github"
 	"github.com/patrickmn/go-cache"
 	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/semaphore"
 )
 
 type Server struct {
 	router   chi.Router
 	log      *logrus.Logger
 	db       *firestore.Client
 	ghClient *github.Client
-	ghMutex  sync.Mutex
 	storage  *s3.Client
 	config   *config.ServerConfig
 	cache    *cache.Cache
-	batchMu  sync.Mutex
+
+	ghSemaphore           *semaphore.Weighted
+	batchArchiveSemaphore *semaphore.Weighted
 }
 
 func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -52,13 +53,15 @@ func (s *Server) indexHandler(w http.ResponseWriter, r *http.Request) {
 func New(log *logrus.Logger, db *firestore.Client, ghClient *github.Client, storage *s3.Client, serverCfg *config.ServerConfig) *Server {
 	router := chi.NewRouter()
 	server := &Server{
-		router:   router,
-		log:      log,
-		db:       db,
-		ghClient: ghClient,
-		storage:  storage,
-		config:   serverCfg,
-		cache:    cache.New(5*time.Minute, 10*time.Minute),
+		router:                router,
+		log:                   log,
+		db:                    db,
+		ghClient:              ghClient,
+		storage:               storage,
+		config:                serverCfg,
+		cache:                 cache.New(5*time.Minute, 10*time.Minute),
+		ghSemaphore:           semaphore.NewWeighted(1),
+		batchArchiveSemaphore: semaphore.NewWeighted(1),
 	}
 	router.Use(middleware.RequestID)
 	router.Use(middleware.RealIP)