diff --git a/AGENTS.md b/AGENTS.md
index be764e5d..c78841a5 100644
--- a/AGENTS.md
+++ b/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.
diff --git a/src/main/java/com/github/wellch4n/oops/application/service/PipelineService.java b/src/main/java/com/github/wellch4n/oops/application/service/PipelineService.java
index 083e704c..dcd16a78 100644
--- a/src/main/java/com/github/wellch4n/oops/application/service/PipelineService.java
+++ b/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
diff --git a/src/main/java/com/github/wellch4n/oops/infrastructure/kubernetes/KubernetesPipelineJobGateway.java b/src/main/java/com/github/wellch4n/oops/infrastructure/kubernetes/KubernetesPipelineJobGateway.java
index 92f032d3..205e93d2 100644
--- a/src/main/java/com/github/wellch4n/oops/infrastructure/kubernetes/KubernetesPipelineJobGateway.java
+++ b/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()
diff --git a/src/main/java/com/github/wellch4n/oops/infrastructure/lock/MysqlNamedLockRegistry.java b/src/main/java/com/github/wellch4n/oops/infrastructure/lock/MysqlNamedLockRegistry.java
new file mode 100644
index 00000000..e37cf60a
--- /dev/null
+++ b/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}).
+ *
+ *
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.
+ *
+ *
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.
+ *
+ *
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 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 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);
+ }
+ }
+}
diff --git a/src/main/java/com/github/wellch4n/oops/infrastructure/lock/NamedLockRegistry.java b/src/main/java/com/github/wellch4n/oops/infrastructure/lock/NamedLockRegistry.java
new file mode 100644
index 00000000..8f7c9869
--- /dev/null
+++ b/src/main/java/com/github/wellch4n/oops/infrastructure/lock/NamedLockRegistry.java
@@ -0,0 +1,33 @@
+package com.github.wellch4n.oops.infrastructure.lock;
+
+import java.util.Set;
+
+/**
+ * Cluster-wide named locks that let several OOPS servers share the scheduled work without stepping on each other:
+ * a pipeline is driven by whichever server holds its lock, and a once-a-minute scan runs on whichever server holds
+ * that scan's lock.
+ *
+ *
A lock belongs to the process that acquired it until that process releases it or dies — there is no lease to
+ * renew and no clock to compare. {@link #tryAcquire} never blocks and is idempotent for the holder, so a job can
+ * call it on every tick to learn whether it is still the one supposed to act.
+ *
+ *
What it is not: a fence. A holder whose connection the database has cut keeps believing it holds the lock until
+ * it next talks to the registry, and another server may already have taken over in the meantime. Anything that
+ * writes on the strength of a lock must therefore still guard the write itself; for pipelines that guard is
+ * {@code PipelineRepository.updateStatusIfMatch}, and the lock only makes the collisions it resolves rare.
+ */
+public interface NamedLockRegistry {
+
+ /**
+ * Takes the lock if nobody else holds it. Returns {@code true} when this process holds the lock afterwards —
+ * including when it already did — and {@code false} when another process holds it or the registry is
+ * unreachable. Never waits.
+ */
+ boolean tryAcquire(String name);
+
+ /** Gives the lock back. A no-op for a lock this process does not hold. */
+ void release(String name);
+
+ /** The names this process currently believes it holds. */
+ Set heldLocks();
+}
diff --git a/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineInstanceScanJob.java b/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineInstanceScanJob.java
index 63c4f76f..a7fbc706 100644
--- a/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineInstanceScanJob.java
+++ b/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineInstanceScanJob.java
@@ -16,12 +16,15 @@
import com.github.wellch4n.oops.application.event.PipelineNotificationType;
import com.github.wellch4n.oops.domain.shared.DeployMode;
import com.github.wellch4n.oops.domain.shared.PipelineStatus;
+import com.github.wellch4n.oops.infrastructure.lock.NamedLockRegistry;
import com.github.wellch4n.oops.application.service.EnvironmentService;
import com.github.wellch4n.oops.shared.exception.EnvironmentUnreachableException;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
@@ -31,6 +34,16 @@
import org.springframework.stereotype.Component;
/**
+ * Carries every in-flight pipeline forward by polling Kubernetes: the build scan turns a finished build Job into a
+ * deploy, and the rollout scan turns a converged StatefulSet into a finished pipeline.
+ *
+ *
With several OOPS servers each pipeline is driven by exactly one of them, and different pipelines by different
+ * servers: a scan only touches a pipeline whose {@value #PIPELINE_LOCK_PREFIX} lock it holds, takes the lock the
+ * first time it sees the pipeline and keeps it until the pipeline finishes. The lock spares the servers from
+ * re-reading each other's Jobs; it is not what keeps the state machine correct — every transition still goes
+ * through {@code updateStatusIfMatch}, which is also what protects a pipeline against a stop issued from another
+ * server while it is being driven here.
+ *
* @author wellCh4n
* @date 2025/7/8
*/
@@ -39,9 +52,16 @@
@Component
public class PipelineInstanceScanJob {
private static final Duration ROLLOUT_TIMEOUT = Duration.ofMinutes(5);
+ private static final Duration DEPLOY_TIMEOUT = Duration.ofMinutes(10);
+ /** Lock name prefix of a pipeline; whichever server holds {@code prefix + id} drives that pipeline. */
+ static final String PIPELINE_LOCK_PREFIX = "oops:pipeline:";
+ private static final List ACTIVE_STATUSES = List.of(
+ PipelineStatus.RUNNING, PipelineStatus.DEPLOYING, PipelineStatus.ROLLING_OUT);
private final AtomicBoolean pipelineJobScanInProgress = new AtomicBoolean(false);
private final AtomicBoolean rolloutScanInProgress = new AtomicBoolean(false);
+ /** When this server first saw each pipeline in DEPLOYING; the clock behind {@link #failStuckDeploy}. */
+ final Map deployingSince = new ConcurrentHashMap<>();
private final ApplicationRepository applicationRepository;
private final PipelineRepository pipelineRepository;
private final EnvironmentService environmentService;
@@ -50,6 +70,7 @@ public class PipelineInstanceScanJob {
private final ArtifactDeploymentExecutor artifactDeploymentExecutor;
private final PipelineStateMachine pipelineStateMachine;
private final ApplicationRuntimeGateway applicationRuntimeGateway;
+ private final NamedLockRegistry lockRegistry;
public PipelineInstanceScanJob(ApplicationRepository applicationRepository,
PipelineRepository pipelineRepository, EnvironmentService environmentService,
@@ -57,7 +78,8 @@ public PipelineInstanceScanJob(ApplicationRepository applicationRepository,
PipelineJobGateway pipelineJobGateway,
ArtifactDeploymentExecutor artifactDeploymentExecutor,
PipelineStateMachine pipelineStateMachine,
- ApplicationRuntimeGateway applicationRuntimeGateway) {
+ ApplicationRuntimeGateway applicationRuntimeGateway,
+ NamedLockRegistry lockRegistry) {
this.applicationRepository = applicationRepository;
this.pipelineRepository = pipelineRepository;
this.environmentService = environmentService;
@@ -66,6 +88,7 @@ public PipelineInstanceScanJob(ApplicationRepository applicationRepository,
this.artifactDeploymentExecutor = artifactDeploymentExecutor;
this.pipelineStateMachine = pipelineStateMachine;
this.applicationRuntimeGateway = applicationRuntimeGateway;
+ this.lockRegistry = lockRegistry;
}
@Scheduled(fixedDelay = 5000)
@@ -77,6 +100,7 @@ public void scanPipelineJobs() {
Map> pipelinesByEnvironment = pipelineRepository
.findAllByStatus(PipelineStatus.RUNNING).stream()
.filter(pipeline -> !pipelineStateMachine.isTerminal(pipeline.getStatus()))
+ .filter(this::claim)
.collect(Collectors.groupingBy(Pipeline::getEnvironment));
for (Map.Entry> entry : pipelinesByEnvironment.entrySet()) {
@@ -108,7 +132,11 @@ public void scanPipelineJobs() {
for (Pipeline pipeline : pipelines) {
try {
- PipelineJobStatus jobStatus = jobStatuses.getOrDefault(pipeline.getId(), PipelineJobStatus.UNKNOWN);
+ if (!jobStatuses.containsKey(pipeline.getId())) {
+ failMissingBuildJob(pipeline);
+ continue;
+ }
+ PipelineJobStatus jobStatus = jobStatuses.get(pipeline.getId());
if (jobStatus == PipelineJobStatus.SUCCEEDED) {
if (DeployMode.MANUAL.equals(pipeline.getDeployMode())) {
pipelineStateMachine.ensureCanTransition(PipelineStatus.RUNNING, PipelineStatus.BUILD_SUCCEEDED);
@@ -156,20 +184,12 @@ public void scanPipelineJobs() {
completeDeployPhase(pipeline);
} else if (jobStatus == PipelineJobStatus.FAILED) {
log.warn("Pipeline build failed: {}", pipeline.getId());
- pipelineStateMachine.ensureCanTransition(PipelineStatus.RUNNING, PipelineStatus.ERROR);
- String message = "镜像构建失败，请查看流水线日志。";
- int updated = pipelineRepository.updateStatusAndMessageIfMatch(
- pipeline.getId(), PipelineStatus.RUNNING, PipelineStatus.ERROR, message
- );
- if (updated > 0) {
- pipeline.markFailed(message);
- eventPublisher.publishEvent(PipelineNotificationEvent.of(
- pipeline, PipelineNotificationType.FAILED, message
- ));
- }
+ failBuild(pipeline, "镜像构建失败，请查看流水线日志。");
}
} catch (Exception exception) {
failPipeline(pipeline, exception);
+ } finally {
+ releaseWhenFinished(pipeline);
}
}
}
@@ -197,6 +217,31 @@ private void failPipeline(Pipeline pipeline, Exception exception) {
pipeline, PipelineNotificationType.FAILED, message
));
}
+ releaseWhenFinished(pipeline);
+ }
+
+ /**
+ * The cluster has no Job for this pipeline any more — deleted by hand, its work namespace cleaned up, or reaped
+ * by the Job's own TTL — and it is not coming back. Left alone the pipeline would stay RUNNING for good, and
+ * with it the application could never be deployed again. A Job that exists but has no status yet is a different
+ * case and is left for the next tick.
+ */
+ private void failMissingBuildJob(Pipeline pipeline) {
+ log.warn("Build job of pipeline {} no longer exists in the cluster", pipeline.getId());
+ failBuild(pipeline, "构建任务已不存在，无法继续。");
+ }
+
+ private void failBuild(Pipeline pipeline, String message) {
+ pipelineStateMachine.ensureCanTransition(PipelineStatus.RUNNING, PipelineStatus.ERROR);
+ int updated = pipelineRepository.updateStatusAndMessageIfMatch(
+ pipeline.getId(), PipelineStatus.RUNNING, PipelineStatus.ERROR, message
+ );
+ if (updated > 0) {
+ pipeline.markFailed(message);
+ eventPublisher.publishEvent(PipelineNotificationEvent.of(
+ pipeline, PipelineNotificationType.FAILED, message
+ ));
+ }
}
/**
@@ -210,8 +255,18 @@ public void scanRollingOutPipelines() {
return;
}
try {
- List rollingOutPipelines = pipelineRepository.findAllByStatus(PipelineStatus.ROLLING_OUT);
- for (Pipeline pipeline : rollingOutPipelines) {
+ // Snapshot before the query: a lock the build scan takes after this point belongs to a pipeline that
+ // may be newer than the list below, and must not be swept as no longer active.
+ Set locksHeldBeforeScan = lockRegistry.heldLocks();
+ List activePipelines = pipelineRepository.findByStatusIn(ACTIVE_STATUSES);
+ for (Pipeline pipeline : activePipelines) {
+ if (pipeline.getStatus() == PipelineStatus.DEPLOYING) {
+ failStuckDeploy(pipeline);
+ continue;
+ }
+ if (pipeline.getStatus() != PipelineStatus.ROLLING_OUT || !claim(pipeline)) {
+ continue;
+ }
try {
Environment environment = environmentService.getEnvironment(pipeline.getEnvironment());
if (environment == null) {
@@ -233,13 +288,85 @@ public void scanRollingOutPipelines() {
// otherwise: still rolling out, leave ROLLING_OUT for the next tick
} catch (Exception exception) {
log.error("Error rolling out pipeline instance {}: {}", pipeline.getId(), exception.getMessage(), exception);
+ } finally {
+ releaseWhenFinished(pipeline);
}
}
+ releaseLocksOfInactivePipelines(locksHeldBeforeScan, activePipelines);
+ forgetDeploysNoLongerActive(activePipelines);
} finally {
rolloutScanInProgress.set(false);
}
}
+ /**
+ * A deploy runs to completion in seconds on the thread that claimed it, so a pipeline that has stayed in
+ * DEPLOYING for {@link #DEPLOY_TIMEOUT} was abandoned: the server driving it died, or its request thread was
+ * killed, between claiming the pipeline and entering the rollout. Nothing else looks at DEPLOYING, so without
+ * this the pipeline would sit there for good and block the application. No column records when a status
+ * changed — the conditional updates bypass entity timestamps — so each server simply remembers when it first saw
+ * the pipeline deploying; a restart only starts that clock again.
+ */
+ private void failStuckDeploy(Pipeline pipeline) {
+ Instant firstSeen = deployingSince.computeIfAbsent(pipeline.getId(), pipelineId -> Instant.now());
+ if (Duration.between(firstSeen, Instant.now()).compareTo(DEPLOY_TIMEOUT) < 0 || !claim(pipeline)) {
+ return;
+ }
+ try {
+ String message = "部署过程中断，未能进入发布阶段。";
+ pipelineStateMachine.ensureCanTransition(PipelineStatus.DEPLOYING, PipelineStatus.ERROR);
+ int updated = pipelineRepository.updateStatusAndMessageIfMatch(
+ pipeline.getId(), PipelineStatus.DEPLOYING, PipelineStatus.ERROR, message);
+ if (updated > 0) {
+ log.warn("Pipeline {} has been deploying for over {}; marking it failed", pipeline.getId(), DEPLOY_TIMEOUT);
+ pipeline.markFailed(message);
+ eventPublisher.publishEvent(PipelineNotificationEvent.of(
+ pipeline, PipelineNotificationType.FAILED, message
+ ));
+ }
+ } finally {
+ releaseWhenFinished(pipeline);
+ }
+ }
+
+ private void forgetDeploysNoLongerActive(List activePipelines) {
+ Set deploying = activePipelines.stream()
+ .filter(pipeline -> pipeline.getStatus() == PipelineStatus.DEPLOYING)
+ .map(Pipeline::getId)
+ .collect(Collectors.toSet());
+ deployingSince.keySet().retainAll(deploying);
+ }
+
+ private boolean claim(Pipeline pipeline) {
+ return lockRegistry.tryAcquire(lockName(pipeline));
+ }
+
+ private void releaseWhenFinished(Pipeline pipeline) {
+ if (pipeline.finished()) {
+ lockRegistry.release(lockName(pipeline));
+ }
+ }
+
+ /**
+ * A pipeline this server drives can still be finished by someone else — a user stopping it from any server —
+ * and then it never comes back through either scan to release its lock. Any lock held before the scan whose
+ * pipeline is no longer active belongs to such a pipeline.
+ */
+ private void releaseLocksOfInactivePipelines(Set locksHeldBeforeScan, List activePipelines) {
+ Set activeLocks = activePipelines.stream()
+ .map(PipelineInstanceScanJob::lockName)
+ .collect(Collectors.toSet());
+ for (String lock : locksHeldBeforeScan) {
+ if (lock.startsWith(PIPELINE_LOCK_PREFIX) && !activeLocks.contains(lock)) {
+ lockRegistry.release(lock);
+ }
+ }
+ }
+
+ private static String lockName(Pipeline pipeline) {
+ return PIPELINE_LOCK_PREFIX + pipeline.getId();
+ }
+
private void succeedRollout(Pipeline pipeline) {
pipelineStateMachine.ensureCanTransition(PipelineStatus.ROLLING_OUT, PipelineStatus.SUCCEEDED);
int updated = pipelineRepository.updateStatusIfMatch(
@@ -270,8 +397,14 @@ private void failRollout(Pipeline pipeline, String message) {
*/
private void completeDeployPhase(Pipeline pipeline) {
pipelineStateMachine.ensureCanTransition(PipelineStatus.DEPLOYING, PipelineStatus.ROLLING_OUT);
- pipelineRepository.updateStatusIfMatch(
+ int updated = pipelineRepository.updateStatusIfMatch(
pipeline.getId(), PipelineStatus.DEPLOYING, PipelineStatus.ROLLING_OUT);
+ if (updated == 0) {
+ // Stopped from some server while the artifact was being applied: the rollout is no longer this
+ // pipeline's to report on, though the workload is updated regardless.
+ 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, "正在等待新版本发布生效…"
diff --git a/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertScanJob.java b/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertScanJob.java
index 26000c4a..7304f1d1 100644
--- a/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertScanJob.java
+++ b/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertScanJob.java
@@ -7,6 +7,7 @@
import com.github.wellch4n.oops.application.service.EnvironmentService;
import com.github.wellch4n.oops.domain.environment.Environment;
import com.github.wellch4n.oops.infrastructure.config.ResourceAlertProperties;
+import com.github.wellch4n.oops.infrastructure.lock.NamedLockRegistry;
import com.github.wellch4n.oops.infrastructure.metrics.MonitoringErrors;
import com.github.wellch4n.oops.infrastructure.persistence.jpa.ApplicationAlertState;
import com.github.wellch4n.oops.infrastructure.persistence.jpa.ApplicationAlertStateRepository;
@@ -47,8 +48,9 @@
* are fanned out across virtual threads. Doing the state lookups inside the fan-out would issue one query per
* concurrent thread and exhaust the connection pool as soon as an installation had a few hundred applications.
*
- *
Single-instance assumption matches {@link PipelineInstanceScanJob} and {@link ScheduledRestartJob}: two servers
- * running this would each evaluate every application and could both notify on the same edge.
+ *
With several servers only the one holding the {@value #LOCK_NAME} lock scans, taken on its first tick and kept
+ * for as long as it lives, as in {@link ScheduledRestartJob}. Two servers evaluating the same edge would each write
+ * it back and both notify; the state rows carry no version to make the second writer lose.
*
*
Conditional rather than a flag checked on each tick, following how the IDE beans opt in: an installation that
* has not asked for alerting does not get a scheduled task at all, instead of one that wakes up every minute to
@@ -76,12 +78,15 @@ private record ProbeOutcome(AlertTarget target,
boolean monitoringMissing) {
}
+ static final String LOCK_NAME = "oops:job:resource-alert";
+
private final AtomicBoolean scanInProgress = new AtomicBoolean(false);
private final ApplicationRuntimeSpecRepository runtimeSpecRepository;
private final ApplicationAlertStateRepository alertStateRepository;
private final EnvironmentService environmentService;
private final ResourceAlertProbe resourceAlertProbe;
private final ResourceAlertProperties properties;
+ private final NamedLockRegistry lockRegistry;
private final ApplicationEventPublisher eventPublisher;
public ResourceAlertScanJob(ApplicationRuntimeSpecRepository runtimeSpecRepository,
@@ -89,12 +94,14 @@ public ResourceAlertScanJob(ApplicationRuntimeSpecRepository runtimeSpecReposito
EnvironmentService environmentService,
ResourceAlertProbe resourceAlertProbe,
ResourceAlertProperties properties,
+ NamedLockRegistry lockRegistry,
ApplicationEventPublisher eventPublisher) {
this.runtimeSpecRepository = runtimeSpecRepository;
this.alertStateRepository = alertStateRepository;
this.environmentService = environmentService;
this.resourceAlertProbe = resourceAlertProbe;
this.properties = properties;
+ this.lockRegistry = lockRegistry;
this.eventPublisher = eventPublisher;
}
@@ -104,6 +111,9 @@ public void scanResourceAlerts() {
return;
}
try {
+ if (!lockRegistry.tryAcquire(LOCK_NAME)) {
+ return;
+ }
List targets = collectTargets();
if (targets.isEmpty()) {
return;
diff --git a/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/ScheduledRestartJob.java b/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/ScheduledRestartJob.java
index 43ecb15a..fc25ce13 100644
--- a/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/ScheduledRestartJob.java
+++ b/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/ScheduledRestartJob.java
@@ -4,6 +4,7 @@
import com.github.wellch4n.oops.application.service.EnvironmentService;
import com.github.wellch4n.oops.domain.environment.Environment;
import com.github.wellch4n.oops.infrastructure.persistence.jpa.ApplicationExpertConfig;
+import com.github.wellch4n.oops.infrastructure.lock.NamedLockRegistry;
import com.github.wellch4n.oops.infrastructure.persistence.jpa.ApplicationExpertConfigRepository;
import com.github.wellch4n.oops.shared.util.CronSchedule;
import java.time.ZonedDateTime;
@@ -29,8 +30,10 @@
* one-minute tick and cause occurrences to be skipped).
*
*
No last-run state is persisted: the minute-aligned tick matches each cron occurrence exactly once, and a
- * restart missed while the server is down is intentionally not backfilled. Single-instance assumption matches
- * {@link PipelineInstanceScanJob}.
+ * restart missed while every server is down is intentionally not backfilled. With several servers only the one
+ * holding the {@value #LOCK_NAME} lock scans: it takes the lock on its first tick and keeps it, so the same server
+ * leads until it goes away and another server's tick finds the lock free. Without that every server would fire
+ * the same restart in the same minute.
*/
@Slf4j
@Component
@@ -39,17 +42,22 @@ public class ScheduledRestartJob {
private record DueRestart(String namespace, String applicationName, Environment environment) {
}
+ static final String LOCK_NAME = "oops:job:scheduled-restart";
+
private final AtomicBoolean scanInProgress = new AtomicBoolean(false);
private final ApplicationExpertConfigRepository expertConfigRepository;
private final EnvironmentService environmentService;
private final ApplicationRuntimeGateway applicationRuntimeGateway;
+ private final NamedLockRegistry lockRegistry;
public ScheduledRestartJob(ApplicationExpertConfigRepository expertConfigRepository,
EnvironmentService environmentService,
- ApplicationRuntimeGateway applicationRuntimeGateway) {
+ ApplicationRuntimeGateway applicationRuntimeGateway,
+ NamedLockRegistry lockRegistry) {
this.expertConfigRepository = expertConfigRepository;
this.environmentService = environmentService;
this.applicationRuntimeGateway = applicationRuntimeGateway;
+ this.lockRegistry = lockRegistry;
}
@Scheduled(cron = "0 * * * * *")
@@ -58,6 +66,9 @@ public void scanScheduledRestarts() {
return;
}
try {
+ if (!lockRegistry.tryAcquire(LOCK_NAME)) {
+ return;
+ }
ZonedDateTime now = ZonedDateTime.now();
List dueRestarts = collectDueRestarts(now);
if (dueRestarts.isEmpty()) {
diff --git a/src/test/java/com/github/wellch4n/oops/application/service/PipelineStopTests.java b/src/test/java/com/github/wellch4n/oops/application/service/PipelineStopTests.java
new file mode 100644
index 00000000..eeb82885
--- /dev/null
+++ b/src/test/java/com/github/wellch4n/oops/application/service/PipelineStopTests.java
@@ -0,0 +1,167 @@
+package com.github.wellch4n.oops.application.service;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.github.wellch4n.oops.application.event.PipelineNotificationEvent;
+import com.github.wellch4n.oops.application.event.PipelineNotificationType;
+import com.github.wellch4n.oops.application.port.ArtifactDeploymentExecutor;
+import com.github.wellch4n.oops.application.port.PipelineJobGateway;
+import com.github.wellch4n.oops.application.port.PipelineLogStreamGateway;
+import com.github.wellch4n.oops.application.port.repository.ApplicationRepository;
+import com.github.wellch4n.oops.application.port.repository.PipelineRepository;
+import com.github.wellch4n.oops.domain.application.Application;
+import com.github.wellch4n.oops.domain.application.ApplicationAccessPolicy;
+import com.github.wellch4n.oops.domain.delivery.DeploymentConcurrencyPolicy;
+import com.github.wellch4n.oops.domain.delivery.Pipeline;
+import com.github.wellch4n.oops.domain.delivery.PipelineStateMachine;
+import com.github.wellch4n.oops.domain.environment.Environment;
+import com.github.wellch4n.oops.domain.shared.Operator;
+import com.github.wellch4n.oops.domain.shared.PipelineStatus;
+import com.github.wellch4n.oops.domain.shared.UserRole;
+import com.github.wellch4n.oops.shared.exception.BizException;
+import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.springframework.context.ApplicationEventPublisher;
+
+/**
+ * Every status change a user triggers is a conditional transition from the status they saw, never an unconditional
+ * save: the scan job may have moved the pipeline on any server between their read and their write, and a save
+ * would silently paper over that — a deploy in progress relabelled as stopped, a stopped pipeline announced as
+ * rolling out.
+ */
+class PipelineStopTests {
+
+ private static final String NAMESPACE = "default";
+ private static final String APP_NAME = "demo";
+ private static final String ENV = "prod";
+ private static final String PIPELINE_ID = "pipeline-id";
+ private static final String OPERATOR = "operator-1";
+
+ private PipelineRepository pipelineRepository;
+ private PipelineJobGateway pipelineJobGateway;
+ private ApplicationEventPublisher eventPublisher;
+ private PipelineService pipelineService;
+
+ @BeforeEach
+ void setUp() {
+ pipelineRepository = mock(PipelineRepository.class);
+ pipelineJobGateway = mock(PipelineJobGateway.class);
+ eventPublisher = mock(ApplicationEventPublisher.class);
+ EnvironmentService environmentService = mock(EnvironmentService.class);
+ ApplicationRepository applicationRepository = mock(ApplicationRepository.class);
+ UserService userService = mock(UserService.class);
+
+ pipelineService = new PipelineService(
+ pipelineRepository,
+ environmentService,
+ applicationRepository,
+ userService,
+ eventPublisher,
+ mock(ArtifactDeploymentExecutor.class),
+ pipelineJobGateway,
+ mock(PipelineLogStreamGateway.class),
+ PipelineStateMachine.getInstance(),
+ new DeploymentConcurrencyPolicy(),
+ new ApplicationAccessPolicy()
+ );
+
+ Environment environment = new Environment();
+ environment.setName(ENV);
+ when(environmentService.getEnvironment(ENV)).thenReturn(environment);
+
+ Application application = new Application();
+ application.setName(APP_NAME);
+ application.setNamespace(NAMESPACE);
+ application.setOwner(OPERATOR);
+ when(applicationRepository.findAggregate(NAMESPACE, APP_NAME)).thenReturn(application);
+ when(userService.findOperatorById(OPERATOR)).thenReturn(new Operator(OPERATOR, UserRole.USER, true));
+ when(pipelineRepository.existsByNamespaceAndApplicationNameAndStatusIn(eq(NAMESPACE), eq(APP_NAME), anyList()))
+ .thenReturn(false);
+ }
+
+ private Pipeline pipelineIn(PipelineStatus status) {
+ Pipeline pipeline = new Pipeline();
+ pipeline.setId(PIPELINE_ID);
+ pipeline.setNamespace(NAMESPACE);
+ pipeline.setApplicationName(APP_NAME);
+ pipeline.setEnvironment(ENV);
+ pipeline.setArtifact("registry.example.com/demo:v1");
+ pipeline.setStatus(status);
+ when(pipelineRepository.findByNamespaceAndApplicationNameAndId(NAMESPACE, APP_NAME, PIPELINE_ID)).thenReturn(pipeline);
+ return pipeline;
+ }
+
+ private List publishedTypes() {
+ ArgumentCaptor