feat: implement repository locking mechanism in webhook and polling handlers (#506)

Add a per-repository lock mechanism to prevent concurrent jobs for the same repository. If a job is already running, new webhook requests receive an immediate HTTP 429 response and poll events get skipped, ensuring only one job per repository runs at a time.

Author: kimdre

cmd/doco-cd/http_handler.go

@@ -398,6 +398,8 @@ func (h *handlerData) WebhookHandler(w http.ResponseWriter, r *http.Request) {
 
 	jobLog.Debug("received webhook event")
 
+	repoName := "unknown"
+
 	// Limit the request body size
 	r.Body = http.MaxBytesReader(w, r.Body, h.appConfig.MaxPayloadSize)
 
@@ -426,11 +428,26 @@ func (h *handlerData) WebhookHandler(w http.ResponseWriter, r *http.Request) {
 			statusCode = http.StatusInternalServerError
 		}
 
-		onError(getRepoName(payload.CloneURL), w, jobLog.With(slog.String("ip", r.RemoteAddr), logger.ErrAttr(err)), errMsg, err.Error(), jobID, statusCode)
+		if payload.CloneURL != "" {
+			repoName = getRepoName(payload.CloneURL)
+		}
+
+		onError(repoName, w, jobLog.With(slog.String("ip", r.RemoteAddr), logger.ErrAttr(err)), errMsg, err.Error(), jobID, statusCode)
+
+		return
+	}
 
+	repoName = getRepoName(payload.CloneURL)
+	lock := getRepoLock(repoName)
+	locked := lock.TryLock()
+
+	if !locked {
+		onError(repoName, w, jobLog, "Another job is still in progress for this repository", nil, jobID, http.StatusTooManyRequests)
 		return
 	}
 
+	defer lock.Unlock()
+
 	HandleEvent(ctx, jobLog, w, h.appConfig, h.dataMountPoint, payload, customTarget, jobID, h.dockerCli, h.dockerClient)
 }
 

cmd/doco-cd/poll_handler.go

@@ -30,8 +30,15 @@ import (
 var (
 	ErrNotManagedByDocoCD = errors.New("stack is not managed by doco-cd")
 	ErrDeploymentConflict = errors.New("another stack with the same name already exists and is not managed by this repository")
+	repoLocks             sync.Map // Map to hold locks for each repository
 )
 
+// getRepoLock retrieves a mutex lock for the given repository name.
+func getRepoLock(repoName string) *sync.Mutex {
+	lockIface, _ := repoLocks.LoadOrStore(repoName, &sync.Mutex{})
+	return lockIface.(*sync.Mutex)
+}
+
 // StartPoll initializes PollJob with the provided configuration and starts the PollHandler goroutine.
 func StartPoll(h *handlerData, pollConfig config.PollConfig, wg *sync.WaitGroup) error {
 	if pollConfig.Interval == 0 {
@@ -61,18 +68,28 @@ func StartPoll(h *handlerData, pollConfig config.PollConfig, wg *sync.WaitGroup)
 
 // PollHandler is a function that handles polling for changes in a repository.
 func (h *handlerData) PollHandler(pollJob *config.PollJob) {
-	logger := h.log.With()
+	repoName := getRepoName(string(pollJob.Config.CloneUrl))
 
+	logger := h.log.With(slog.String("repository", repoName))
 	logger.Debug("Start poll handler")
 
+	lock := getRepoLock(repoName)
+
 	for {
 		if pollJob.LastRun == 0 || time.Now().Unix() >= pollJob.NextRun {
-			repoName := getRepoName(string(pollJob.Config.CloneUrl))
-			logger.Debug("Running poll for repository", slog.String("repoName", repoName))
+			locked := lock.TryLock()
 
-			err := RunPoll(context.Background(), pollJob.Config, h.appConfig, h.dataMountPoint, h.dockerCli, h.dockerClient, logger)
-			if err != nil {
-				prometheus.PollErrors.WithLabelValues(repoName).Inc()
+			if !locked {
+				h.log.Info("Another poll job is still in progress, skipping this run")
+			} else {
+				logger.Debug("Start poll job")
+
+				err := RunPoll(context.Background(), pollJob.Config, h.appConfig, h.dataMountPoint, h.dockerCli, h.dockerClient, logger)
+				if err != nil {
+					prometheus.PollErrors.WithLabelValues(repoName).Inc()
+				}
+
+				lock.Unlock()
 			}
 
 			pollJob.NextRun = time.Now().Unix() + int64(pollJob.Config.Interval)
@@ -91,9 +108,7 @@ func RunPoll(ctx context.Context, pollConfig config.PollConfig, appConfig *confi
 	cloneUrl := string(pollConfig.CloneUrl)
 	jobID := uuid.Must(uuid.NewRandom()).String()
 	repoName := getRepoName(cloneUrl)
-	jobLog := logger.With(
-		slog.String("repository", repoName),
-		slog.String("job_id", jobID))
+	jobLog := logger.With(slog.String("job_id", jobID))
 
 	if strings.Contains(repoName, "..") {
 		jobLog.Error("invalid repository name, contains '..'")