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. The container names
 
 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/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()

src/main/java/com/github/wellch4n/oops/infrastructure/lock/MysqlNamedLockRegistry.java

@@ -0,0 +1,180 @@
+package com.github.wellch4n.oops.infrastructure.lock;
+
+import jakarta.annotation.PreDestroy;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.LinkedHashSet;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * {@link NamedLockRegistry} over MySQL's user-level locks ({@code GET_LOCK} / {@code RELEASE_LOCK}).
+ *
+ * <p>A user-level lock lives exactly as long as the session that took it, which is the whole point: when this
+ * process dies, the TCP connection drops and MySQL frees every lock it held, with no expiry to tune and no heartbeat
+ * to send. That also dictates the one unusual thing here — the registry keeps a single dedicated connection open
+ * for its entire life, outside the Hikari pool. A pooled connection would either be returned (releasing the locks)
+ * or held forever (shrinking the pool by one and tripping leak detection), and the pool's own retirement of
+ * connections would silently drop the locks.
+ *
+ * <p>Losing that connection is the failure mode to understand. Whatever the cause, the server-side locks are gone
+ * with it, so the registry forgets everything it held; the next {@link #tryAcquire} reconnects and starts over, and
+ * until it succeeds every call answers {@code false}. Fail-closed on purpose: with the lock unreachable no server
+ * drives anything, which is recoverable, whereas two servers driving the same thing is what the lock exists to
+ * prevent. A 30-second validation keeps the idle session clear of {@code wait_timeout} and notices a cut connection
+ * before the next tick does.
+ *
+ * <p>All access is serialised on the registry: a JDBC connection is not safe for concurrent use, and the scheduled
+ * jobs that call in run on a pool of threads.
+ */
+@Slf4j
+@Component
+public class MysqlNamedLockRegistry implements NamedLockRegistry {
+
+    /** MySQL rejects user-level lock names longer than this. */
+    private static final int MAX_NAME_LENGTH = 64;
+    private static final int VALIDATION_TIMEOUT_SECONDS = 5;
+
+    private final DataSourceProperties dataSourceProperties;
+    private final Set<String> held = new LinkedHashSet<>();
+    private Connection connection;
+    private boolean connectionFailureLogged;
+
+    public MysqlNamedLockRegistry(DataSourceProperties dataSourceProperties) {
+        this.dataSourceProperties = dataSourceProperties;
+    }
+
+    @Override
+    public synchronized boolean tryAcquire(String name) {
+        requireValidName(name);
+        if (held.contains(name)) {
+            return true;
+        }
+        Connection lockConnection = connection();
+        if (lockConnection == null) {
+            return false;
+        }
+        try (PreparedStatement statement = lockConnection.prepareStatement("SELECT GET_LOCK(?, 0)")) {
+            statement.setString(1, name);
+            try (ResultSet resultSet = statement.executeQuery()) {
+                // 1 = acquired, 0 = held by another session, NULL = error; only 1 means it is ours.
+                boolean acquired = resultSet.next() && resultSet.getInt(1) == 1;
+                if (acquired) {
+                    held.add(name);
+                }
+                return acquired;
+            }
+        } catch (SQLException exception) {
+            dropConnection("acquiring lock " + name, exception.getMessage());
+            return false;
+        }
+    }
+
+    @Override
+    public synchronized void release(String name) {
+        if (!held.remove(name) || connection == null) {
+            return;
+        }
+        try (PreparedStatement statement = connection.prepareStatement("SELECT RELEASE_LOCK(?)")) {
+            statement.setString(1, name);
+            statement.executeQuery().close();
+        } catch (SQLException exception) {
+            dropConnection("releasing lock " + name, exception.getMessage());
+        }
+    }
+
+    @Override
+    public synchronized Set<String> heldLocks() {
+        return Set.copyOf(held);
+    }
+
+    /**
+     * Pings the lock session so MySQL's {@code wait_timeout} never reaps it while idle, and forgets the held locks
+     * as soon as the connection turns out to be gone rather than at the next acquire.
+     */
+    @Scheduled(fixedDelay = 30_000)
+    public synchronized void keepAlive() {
+        if (connection == null) {
+            return;
+        }
+        try {
+            if (!connection.isValid(VALIDATION_TIMEOUT_SECONDS)) {
+                dropConnection("validating the lock connection", "connection is no longer valid");
+            }
+        } catch (SQLException exception) {
+            dropConnection("validating the lock connection", exception.getMessage());
+        }
+    }
+
+    @PreDestroy
+    public synchronized void close() {
+        if (connection == null) {
+            return;
+        }
+        // Closing the session releases every lock it holds; another server picks the work up on its next tick.
+        try {
+            connection.close();
+        } catch (SQLException exception) {
+            log.debug("Closing the lock connection failed: {}", exception.getMessage());
+        }
+        connection = null;
+        held.clear();
+    }
+
+    private Connection connection() {
+        if (connection != null) {
+            return connection;
+        }
+        try {
+            Connection opened = DriverManager.getConnection(
+                    dataSourceProperties.determineUrl(),
+                    dataSourceProperties.determineUsername(),
+                    dataSourceProperties.determinePassword());
+            opened.setAutoCommit(true);
+            connection = opened;
+            if (connectionFailureLogged) {
+                log.info("Lock connection re-established; this server takes part in scheduled work again");
+                connectionFailureLogged = false;
+            }
+            return opened;
+        } catch (SQLException exception) {
+            if (!connectionFailureLogged) {
+                log.warn("Cannot open the lock connection; scheduled work is paused on this server until it is back: {}",
+                        exception.getMessage());
+                connectionFailureLogged = true;
+            }
+            return null;
+        }
+    }
+
+    private void dropConnection(String activity, String reason) {
+        log.warn("Lock connection lost while {}; the {} lock(s) this server held are gone with it: {}",
+                activity, held.size(), reason);
+        held.clear();
+        Connection lost = connection;
+        connection = null;
+        if (lost != null) {
+            try {
+                lost.close();
+            } catch (SQLException exception) {
+                log.debug("Closing the lost lock connection failed: {}", exception.getMessage());
+            }
+        }
+    }
+
+    private static void requireValidName(String name) {
+        if (StringUtils.isBlank(name)) {
+            throw new IllegalArgumentException("Lock name is required");
+        }
+        if (name.length() > MAX_NAME_LENGTH) {
+            throw new IllegalArgumentException("Lock name exceeds " + MAX_NAME_LENGTH + " characters: " + name);
+        }
+    }
+}