feat(scheduler): drive each pipeline from one server with MySQL named locks

AGENTS.md

@@ -24,6 +24,7 @@ The backend follows a **DDD layered architecture** under `com.github.wellch4n.oo
   - `infrastructure/external/feishu/` — Feishu (Lark) OAuth + messaging
   - `infrastructure/objectstorage/` — S3-compatible storage for ZIP source uploads and static assets
   - `infrastructure/scheduler/` — `@Scheduled` jobs (e.g. `PipelineInstanceScanJob`)
+  - `infrastructure/lock/` — `NamedLockRegistry` and its MySQL `GET_LOCK` implementation, which lets several backend servers share the scheduled jobs (see **Multi-server coordination**)
   - `infrastructure/config/` — Spring config beans, `@ConfigurationProperties` classes, `SpringContext` static accessor, security filters (`JwtAuthFilter`, `OpenApiAuthFilter`), and startup initializers (`UserInitializer` seeds the default admin)
 - **`shared/`**: Cross-cutting utilities and exceptions (`BizException`).
 
@@ -172,7 +173,14 @@ Build pipelines run as Kubernetes Jobs with init containers:
 
 Two source types exist: `GIT` (default) and `ZIP`. ZIP uploads use presigned S3 URLs via `BuildSourceObjectStorageService` — the frontend gets a presigned PUT URL from `POST .../deployments/source-upload`, uploads the file, then triggers the pipeline. ZIP builds use `oops.pipeline.image.zip` (defaults to `alpine/curl:8.17.0`) to download the archive.
 
-Pipeline build logs are served over SSE, one container at a time (see **Pipeline log streaming** below). A `@Scheduled(fixedRate=5000)` job (`PipelineInstanceScanJob`) polls K8s for build completion and rollout convergence. Pipeline state transitions use optimistic locking: `PipelineRepository.updateStatusIfMatch()` does a conditional UPDATE and returns row count (0 = lost the race).
+Pipeline build logs are served over SSE, one container at a time (see **Pipeline log streaming** below). A `@Scheduled(fixedRate=5000)` job (`PipelineInstanceScanJob`) polls K8s for build completion and rollout convergence. Pipeline state transitions use optimistic locking: `PipelineRepository.updateStatusIfMatch()` does a conditional UPDATE and returns row count (0 = lost the race). **Every** status write goes through it, user-triggered ones included — `stopPipeline` transitions from the status the caller read and fails with "state changed concurrently" when that read is stale, and a lost `DEPLOYING → ROLLING_OUT` claim (the pipeline was stopped while its artifact was applied) is logged and not announced as rolling out. Two ways a pipeline could otherwise stick forever and block its application are recovered by the scan job: a RUNNING pipeline whose build Job the cluster no longer has (deleted, namespace cleaned, TTL-reaped) is failed at once, and a pipeline that has stayed DEPLOYING for 10 minutes (its driver died mid-deploy) is failed by whichever server can claim its lock — measured from when that server first saw it deploying, since no column records status-change time and the conditional updates bypass entity timestamps.
+
+**Multi-server coordination**: Several backend servers can run at once, and all of them run every `@Scheduled` job; what stops them from doing the same work twice is `NamedLockRegistry` (`infrastructure/lock/`), implemented over MySQL user-level locks (`GET_LOCK` / `RELEASE_LOCK`) on **one dedicated connection per server that is opened outside the Hikari pool and kept for the life of the process**. A lock lives exactly as long as the session that took it, so a dead server's locks are freed the moment its TCP connection drops — no lease to renew, no clock involved. Two granularities:
+
+- **Per pipeline** — `PipelineInstanceScanJob` only touches a pipeline whose `oops:pipeline:{id}` lock it holds, takes it the first time it sees the pipeline (build scan or rollout scan) and keeps it until the pipeline finishes, so one server drives a pipeline from build through rollout while other servers drive other pipelines. A lock whose pipeline was finished elsewhere (a user stopping it from any server) is swept by the rollout scan: it snapshots the held locks *before* querying the active pipelines and releases any snapshot lock that is no longer active — the snapshot ordering is what keeps a lock the build scan takes concurrently from being swept.
+- **Per job, sticky** — `ScheduledRestartJob` and `ResourceAlertScanJob` take `oops:job:scheduled-restart` / `oops:job:resource-alert` on their first tick and never release, so one server leads each scan until it goes away; a scan that released its lock after each tick would let a second server's tick in the same minute fire the same restart or notification again.
+
+The lock is **not a fence**: a server whose lock connection MySQL cut keeps believing it holds its locks until it next talks to the registry. That is why every pipeline transition still goes through `updateStatusIfMatch()` — the lock makes collisions rare, the CAS makes them harmless. The registry is fail-closed: with its connection down every `tryAcquire` answers false and that server simply drives nothing until it reconnects (it retries on the next call and validates the idle session every 30s).
 
 **Duplicate deploy guard**: Before starting any pipeline, manual deploy, or rollback, `DeploymentConcurrencyPolicy.activePipelineStatuses()` checks for in-flight pipelines (status `RUNNING`, `DEPLOYING`, or `ROLLING_OUT`) through `PipelineRepository.existsByNamespaceAndApplicationNameAndStatusIn()`. If one exists, a `BizException("Application is being deployed")` is thrown. This is separate from optimistic locking — it's a pre-check to prevent duplicate concurrent deployments.
 

src/main/java/com/github/wellch4n/oops/application/service/PipelineService.java

@@ -26,6 +26,7 @@
 import com.github.wellch4n.oops.application.dto.PipelineDto;
 import java.util.*;
 import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
@@ -35,6 +36,7 @@
  * @date 2025/7/28
  */
 
+@Slf4j
 @Service
 public class PipelineService {
 
@@ -211,12 +213,14 @@ public Boolean deployPipeline(String namespace, String applicationName, String i
         } catch (Exception exception) {
             pipelineStateMachine.ensureCanTransition(PipelineStatus.DEPLOYING, PipelineStatus.ERROR);
             String message = StringUtils.defaultIfBlank(exception.getMessage(), "发布任务执行失败，请查看日志。");
-            pipelineRepository.updateStatusAndMessageIfMatch(
+            int failed = pipelineRepository.updateStatusAndMessageIfMatch(
                     pipeline.getId(), PipelineStatus.DEPLOYING, PipelineStatus.ERROR, message);
-            pipeline.markFailed(message);
-            eventPublisher.publishEvent(PipelineNotificationEvent.of(
-                    pipeline, PipelineNotificationType.FAILED, message
-            ));
+            if (failed > 0) {
+                pipeline.markFailed(message);
+                eventPublisher.publishEvent(PipelineNotificationEvent.of(
+                        pipeline, PipelineNotificationType.FAILED, message
+                ));
+            }
             throw new BizException("Deploy failed: " + exception.getMessage(), exception);
         }
         return true;
@@ -269,12 +273,14 @@ public String rollback(String namespace, String applicationName, String targetPi
         } catch (Exception exception) {
             pipelineStateMachine.ensureCanTransition(PipelineStatus.DEPLOYING, PipelineStatus.ERROR);
             String message = StringUtils.defaultIfBlank(exception.getMessage(), "回滚任务执行失败，请查看日志。");
-            pipelineRepository.updateStatusAndMessageIfMatch(
+            int failed = pipelineRepository.updateStatusAndMessageIfMatch(
                     rollbackPipeline.getId(), PipelineStatus.DEPLOYING, PipelineStatus.ERROR, message);
-            rollbackPipeline.markFailed(message);
-            eventPublisher.publishEvent(PipelineNotificationEvent.of(
-                    rollbackPipeline, PipelineNotificationType.FAILED, message
-            ));
+            if (failed > 0) {
+                rollbackPipeline.markFailed(message);
+                eventPublisher.publishEvent(PipelineNotificationEvent.of(
+                        rollbackPipeline, PipelineNotificationType.FAILED, message
+                ));
+            }
             throw new BizException("Rollback failed: " + exception.getMessage(), exception);
         }
         return rollbackPipeline.getId();
@@ -286,23 +292,23 @@ public Boolean stopPipeline(String namespace, String applicationName, String id,
             throw new BizException("Pipeline not found");
         }
         requireOperableApplication(namespace, applicationName, operatorUserId);
-        pipelineStateMachine.ensureCanTransition(pipeline.getStatus(), PipelineStatus.STOPPED);
-
-        if (pipeline.getStatus() == PipelineStatus.BUILD_SUCCEEDED) {
-            pipeline.stop();
-            pipelineRepository.save(pipeline);
-            eventPublisher.publishEvent(PipelineNotificationEvent.of(
-                    pipeline, PipelineNotificationType.STOPPED, "发布任务已被手动停止。"
-            ));
-            return true;
-        }
+        PipelineStatus current = pipeline.getStatus();
+        pipelineStateMachine.ensureCanTransition(current, PipelineStatus.STOPPED);
 
-        String environmentName = pipeline.getEnvironment();
-        Environment environment = requireEnvironment(environmentName);
+        // A built pipeline has no Job left to stop; every other stoppable status may still be building.
+        if (current != PipelineStatus.BUILD_SUCCEEDED) {
+            Environment environment = requireEnvironment(pipeline.getEnvironment());
+            pipelineJobGateway.stop(environment, pipeline.getName());
+        }
 
-        pipelineJobGateway.stop(environment, pipeline.getName());
+        // The status read above can already be stale: the scan job may have carried a RUNNING pipeline into
+        // DEPLOYING on any server since. Only a transition from the status the caller saw may stop it — an
+        // unconditional save would overwrite the deploy in progress with STOPPED and leave it running anyway.
+        int stopped = pipelineRepository.updateStatusIfMatch(pipeline.getId(), current, PipelineStatus.STOPPED);
+        if (stopped == 0) {
+            throw new BizException("Pipeline state changed concurrently, please refresh");
+        }
         pipeline.stop();
-        pipelineRepository.save(pipeline);
         eventPublisher.publishEvent(PipelineNotificationEvent.of(
                 pipeline, PipelineNotificationType.STOPPED, "发布任务已被手动停止。"
         ));
@@ -315,8 +321,15 @@ public Boolean stopPipeline(String namespace, String applicationName, String id,
      */
     private void completeDeployPhase(Pipeline pipeline, String rollingOutDetail) {
         pipelineStateMachine.ensureCanTransition(PipelineStatus.DEPLOYING, PipelineStatus.ROLLING_OUT);
-        pipelineRepository.updateStatusIfMatch(
+        int updated = pipelineRepository.updateStatusIfMatch(
                 pipeline.getId(), PipelineStatus.DEPLOYING, PipelineStatus.ROLLING_OUT);
+        if (updated == 0) {
+            // The pipeline was moved while its artifact was being applied — a stop is the only legal way — so the
+            // rollout is no longer this pipeline's to report on. The workload is updated regardless: a stop cannot
+            // take back an artifact that has already been applied.
+            log.info("Pipeline {} left DEPLOYING while its artifact was applied; not entering rollout", pipeline.getId());
+            return;
+        }
         pipeline.markRollingOut();
         eventPublisher.publishEvent(PipelineNotificationEvent.of(
                 pipeline, PipelineNotificationType.ROLLING_OUT, rollingOutDetail

src/main/java/com/github/wellch4n/oops/infrastructure/config/SecurityConfiguration.java

@@ -1,5 +1,6 @@
 package com.github.wellch4n.oops.infrastructure.config;
 
+import jakarta.servlet.DispatcherType;
 import jakarta.servlet.http.HttpServletResponse;
 import java.util.List;
 import org.springframework.context.annotation.Bean;
@@ -43,6 +44,14 @@ public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
                                 response.sendError(HttpServletResponse.SC_FORBIDDEN))
                 )
                 .authorizeHttpRequests(auth -> auth
+                        // The SSE endpoints (pipeline steps and logs, pod statuses) complete on an ASYNC
+                        // dispatch. The JWT filter, a OncePerRequestFilter, does not run again on that
+                        // dispatch and nothing carries the authentication over, so without this rule the
+                        // authorization filter denies the dispatch after the whole stream has been sent:
+                        // an error in the log for every stream, and a chunked response cut short of its
+                        // terminator, which any client but a browser's EventSource reports as an error.
+                        // The original request was authorized when it came in; the dispatch is the same one.
+                        .dispatcherTypeMatchers(DispatcherType.ASYNC).permitAll()
                         .requestMatchers("/api/auth/**").permitAll()
                         .requestMatchers("/api/auth/feishu/**").permitAll()
                         .requestMatchers("/api/health").permitAll()

src/main/java/com/github/wellch4n/oops/infrastructure/kubernetes/KubernetesPipelineJobGateway.java

@@ -73,10 +73,14 @@ private PipelineJobStatus toStatus(Job job) {
     @Override
     public void stop(Environment environment, String jobName) {
         var client = clientPool.get(environment.getKubernetesApiServer());
-        client.batch().v1().jobs()
+        var jobResource = client.batch().v1().jobs()
                 .inNamespace(environment.getWorkNamespace())
-                .withName(jobName)
-                .edit(job -> new JobBuilder(job)
+                .withName(jobName);
+        if (jobResource.get() == null) {
+            // Already gone — reaped by its TTL or deleted by hand — so there is nothing left to stop.
+            return;
+        }
+        jobResource.edit(job -> new JobBuilder(job)
                         .editSpec()
                         .withSuspend(true)
                         .endSpec()