From 24961944ab3404d652afd72539ccbe861d412a2d Mon Sep 17 00:00:00 2001 From: wellCh4n Date: Fri, 4 Sep 2026 19:05:15 +0800 Subject: [PATCH 1/2] feat(scheduler): share the scheduled jobs between servers with MySQL named locks Every backend server runs every @Scheduled job. The pipeline scans were safe only because each transition is a conditional UPDATE, and the restart and alert scans were not safe at all: two servers fired the same scheduled restart and sent the same alert in the same minute. Add NamedLockRegistry over MySQL user-level locks (GET_LOCK / RELEASE_LOCK) on one dedicated connection per server, opened outside the Hikari pool and kept for the life of the process. A lock lives as long as its session, so a dead server's locks are freed the moment its connection drops, with no lease to renew and no clock involved. The registry is fail-closed: with its connection down a server drives nothing until it reconnects. The pipeline scans take one lock per pipeline (oops:pipeline:{id}), so a pipeline is driven by one server from build through rollout while other servers drive other pipelines. The rollout scan sweeps locks whose pipeline was finished elsewhere, snapshotting the held locks before it queries so a lock the build scan takes concurrently is never swept. The restart and alert scans take one sticky lock each on their first tick and keep it, so one server leads until it goes away. The lock is not a fence - a server whose session MySQL cut keeps believing in its locks until it next talks to the registry - which is why every pipeline transition still goes through updateStatusIfMatch. Co-Authored-By: Claude Fable 5.1 --- AGENTS.md | 8 + .../lock/MysqlNamedLockRegistry.java | 180 ++++++++++++++++++ .../lock/NamedLockRegistry.java | 33 ++++ .../scheduler/PipelineInstanceScanJob.java | 68 ++++++- .../scheduler/ResourceAlertScanJob.java | 14 +- .../scheduler/ScheduledRestartJob.java | 17 +- .../lock/MysqlNamedLockRegistryTests.java | 84 ++++++++ .../PipelineVerificationScanTests.java | 103 +++++++++- .../scheduler/ResourceAlertOptInTests.java | 4 +- .../scheduler/ResourceAlertScanJobTests.java | 20 +- .../scheduler/ScheduledRestartJobTests.java | 79 ++++++++ 11 files changed, 593 insertions(+), 17 deletions(-) create mode 100644 src/main/java/com/github/wellch4n/oops/infrastructure/lock/MysqlNamedLockRegistry.java create mode 100644 src/main/java/com/github/wellch4n/oops/infrastructure/lock/NamedLockRegistry.java create mode 100644 src/test/java/com/github/wellch4n/oops/infrastructure/lock/MysqlNamedLockRegistryTests.java create mode 100644 src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ScheduledRestartJobTests.java diff --git a/AGENTS.md b/AGENTS.md index be764e5d..5898fa29 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`). @@ -174,6 +175,13 @@ Two source types exist: `GIT` (default) and `ZIP`. ZIP uploads use presigned S3 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). +**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. **Pipeline notifications**: A Spring event system sends notifications at each pipeline state transition. `PipelineNotificationEvent` is published and `PipelineNotificationListener` (`@EventListener`) formats messages sent to the operator via `ExternalMessageService` (Feishu). Notification types: `CREATED`, `BUILD_SUCCEEDED`, `DEPLOYING`, `ROLLING_OUT`, `SUCCEEDED`, `FAILED`, `STOPPED`. 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..03909b15 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,14 @@ 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.atomic.AtomicBoolean; import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; @@ -31,6 +33,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,6 +51,10 @@ @Component public class PipelineInstanceScanJob { private static final Duration ROLLOUT_TIMEOUT = Duration.ofMinutes(5); + /** 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); @@ -50,6 +66,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 +74,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 +84,7 @@ public PipelineInstanceScanJob(ApplicationRepository applicationRepository, this.artifactDeploymentExecutor = artifactDeploymentExecutor; this.pipelineStateMachine = pipelineStateMachine; this.applicationRuntimeGateway = applicationRuntimeGateway; + this.lockRegistry = lockRegistry; } @Scheduled(fixedDelay = 5000) @@ -77,6 +96,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()) { @@ -170,6 +190,8 @@ public void scanPipelineJobs() { } } catch (Exception exception) { failPipeline(pipeline, exception); + } finally { + releaseWhenFinished(pipeline); } } } @@ -197,6 +219,7 @@ private void failPipeline(Pipeline pipeline, Exception exception) { pipeline, PipelineNotificationType.FAILED, message )); } + releaseWhenFinished(pipeline); } /** @@ -210,8 +233,14 @@ 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.ROLLING_OUT || !claim(pipeline)) { + continue; + } try { Environment environment = environmentService.getEnvironment(pipeline.getEnvironment()); if (environment == null) { @@ -233,13 +262,46 @@ 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); } finally { rolloutScanInProgress.set(false); } } + 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( 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/infrastructure/lock/MysqlNamedLockRegistryTests.java b/src/test/java/com/github/wellch4n/oops/infrastructure/lock/MysqlNamedLockRegistryTests.java new file mode 100644 index 00000000..a25f05aa --- /dev/null +++ b/src/test/java/com/github/wellch4n/oops/infrastructure/lock/MysqlNamedLockRegistryTests.java @@ -0,0 +1,84 @@ +package com.github.wellch4n.oops.infrastructure.lock; + +import static org.junit.jupiter.api.Assertions.assertEquals; +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 java.util.Set; +import java.util.UUID; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties; +import org.springframework.boot.test.context.SpringBootTest; + +/** + * Runs against the configured MySQL like the rest of the Spring-backed tests: user-level locks are a server-side + * behaviour, so two registries — two sessions, as two OOPS servers would be — are the only honest way to see that + * a lock is exclusive and that closing a session hands its locks back. + */ +@SpringBootTest +class MysqlNamedLockRegistryTests { + + @Autowired + private DataSourceProperties dataSourceProperties; + + private MysqlNamedLockRegistry firstServer; + private MysqlNamedLockRegistry secondServer; + private String lockName; + + @BeforeEach + void setUp() { + firstServer = new MysqlNamedLockRegistry(dataSourceProperties); + secondServer = new MysqlNamedLockRegistry(dataSourceProperties); + lockName = "oops:test:" + UUID.randomUUID(); + } + + @AfterEach + void tearDown() { + firstServer.close(); + secondServer.close(); + } + + @Test + void lockBelongsToOneServerUntilReleased() { + assertTrue(firstServer.tryAcquire(lockName)); + assertTrue(firstServer.tryAcquire(lockName), "re-asking for a held lock keeps answering yes"); + assertFalse(secondServer.tryAcquire(lockName), "another server does not get a held lock"); + assertEquals(Set.of(lockName), firstServer.heldLocks()); + assertEquals(Set.of(), secondServer.heldLocks()); + + firstServer.release(lockName); + + assertEquals(Set.of(), firstServer.heldLocks()); + assertTrue(secondServer.tryAcquire(lockName), "a released lock is free for the next server"); + assertFalse(firstServer.tryAcquire(lockName)); + } + + /** What happens when a server dies: its session goes, and with it every lock it held. */ + @Test + void closingTheSessionFreesItsLocks() { + assertTrue(firstServer.tryAcquire(lockName)); + assertFalse(secondServer.tryAcquire(lockName)); + + firstServer.close(); + + assertTrue(secondServer.tryAcquire(lockName)); + assertEquals(Set.of(), firstServer.heldLocks()); + } + + @Test + void releasingALockNeverHeldIsANoOp() { + firstServer.release(lockName); + + assertTrue(secondServer.tryAcquire(lockName)); + } + + @Test + void rejectsNamesMysqlWouldReject() { + assertThrows(IllegalArgumentException.class, () -> firstServer.tryAcquire(" ")); + assertThrows(IllegalArgumentException.class, () -> firstServer.tryAcquire("x".repeat(65))); + } +} diff --git a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineVerificationScanTests.java b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineVerificationScanTests.java index d3d917ab..07de3add 100644 --- a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineVerificationScanTests.java +++ b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineVerificationScanTests.java @@ -1,6 +1,7 @@ package com.github.wellch4n.oops.infrastructure.scheduler; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -19,11 +20,13 @@ import com.github.wellch4n.oops.domain.delivery.PipelineStateMachine; import com.github.wellch4n.oops.domain.environment.Environment; import com.github.wellch4n.oops.domain.shared.PipelineStatus; +import com.github.wellch4n.oops.infrastructure.lock.NamedLockRegistry; import com.github.wellch4n.oops.shared.exception.EnvironmentUnreachableException; import java.net.SocketTimeoutException; import java.time.Instant; import java.util.List; import java.util.Map; +import java.util.Set; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -33,7 +36,9 @@ * Exercises the scan job's decision logic. For ROLLING_OUT: a converged rollout succeeds, a fatal pod state fails * fast, a rollout that has stayed not-ready too long times out, and an in-progress rollout is left untouched. For * RUNNING: every pipeline of an environment is polled in a single request, and a poll that never reached the - * cluster leaves those pipelines alone rather than failing their builds. + * cluster leaves those pipelines alone rather than failing their builds. Across servers: a pipeline whose lock + * another server holds is left to that server, a lock is released as soon as its pipeline finishes, and a lock + * whose pipeline was finished elsewhere is swept. */ class PipelineVerificationScanTests { @@ -48,6 +53,7 @@ class PipelineVerificationScanTests { private ApplicationRuntimeGateway applicationRuntimeGateway; private PipelineJobGateway pipelineJobGateway; private ApplicationEventPublisher eventPublisher; + private NamedLockRegistry lockRegistry; private PipelineInstanceScanJob scanJob; @BeforeEach @@ -59,6 +65,7 @@ void setUp() { eventPublisher = Mockito.mock(ApplicationEventPublisher.class); pipelineJobGateway = Mockito.mock(PipelineJobGateway.class); ArtifactDeploymentExecutor artifactDeploymentExecutor = Mockito.mock(ArtifactDeploymentExecutor.class); + lockRegistry = Mockito.mock(NamedLockRegistry.class); scanJob = new PipelineInstanceScanJob( applicationRepository, @@ -68,11 +75,16 @@ void setUp() { pipelineJobGateway, artifactDeploymentExecutor, PipelineStateMachine.getInstance(), - applicationRuntimeGateway + applicationRuntimeGateway, + lockRegistry ); // No RUNNING pipelines by default; the build branch is a no-op for the rollout tests. when(pipelineRepository.findAllByStatus(PipelineStatus.RUNNING)).thenReturn(List.of()); + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of()); + // This server holds no lock yet and wins every lock it asks for, as on a single-server installation. + when(lockRegistry.tryAcquire(any())).thenReturn(true); + when(lockRegistry.heldLocks()).thenReturn(Set.of()); Environment environment = new Environment(); environment.setName(ENV); @@ -91,7 +103,7 @@ private Pipeline rollingOutPipeline() { @Test void convergedRolloutMarksSucceeded() { - when(pipelineRepository.findAllByStatus(PipelineStatus.ROLLING_OUT)) + when(pipelineRepository.findByStatusIn(anyList())) .thenReturn(List.of(rollingOutPipeline())); when(applicationRuntimeGateway.getDeploymentHealth(any(), eq(NAMESPACE), eq(APP_NAME))) .thenReturn(new DeploymentHealth(false, true, 1, 1, null, null)); @@ -105,7 +117,7 @@ void convergedRolloutMarksSucceeded() { @Test void fatalPodStateMarksErrorBeforeDeadline() { - when(pipelineRepository.findAllByStatus(PipelineStatus.ROLLING_OUT)) + when(pipelineRepository.findByStatusIn(anyList())) .thenReturn(List.of(rollingOutPipeline())); when(applicationRuntimeGateway.getDeploymentHealth(any(), eq(NAMESPACE), eq(APP_NAME))) .thenReturn(new DeploymentHealth(false, false, 1, 0, "ImagePullBackOff (demo-0)", Instant.now())); @@ -122,7 +134,7 @@ void fatalPodStateMarksErrorBeforeDeadline() { @Test void notReadyLongerThanRolloutTimeoutMarksError() { - when(pipelineRepository.findAllByStatus(PipelineStatus.ROLLING_OUT)) + when(pipelineRepository.findByStatusIn(anyList())) .thenReturn(List.of(rollingOutPipeline())); when(applicationRuntimeGateway.getDeploymentHealth(any(), eq(NAMESPACE), eq(APP_NAME))) .thenReturn(new DeploymentHealth(false, false, 2, 1, null, Instant.now().minusSeconds(301))); @@ -137,7 +149,7 @@ void notReadyLongerThanRolloutTimeoutMarksError() { @Test void healthQueryErrorLeavesRollingOutUntouched() { - when(pipelineRepository.findAllByStatus(PipelineStatus.ROLLING_OUT)) + when(pipelineRepository.findByStatusIn(anyList())) .thenReturn(List.of(rollingOutPipeline())); when(applicationRuntimeGateway.getDeploymentHealth(any(), eq(NAMESPACE), eq(APP_NAME))) .thenThrow(new IllegalStateException("Kubernetes API unavailable")); @@ -150,7 +162,7 @@ void healthQueryErrorLeavesRollingOutUntouched() { @Test void inProgressRolloutLeavesRollingOutUntouched() { - when(pipelineRepository.findAllByStatus(PipelineStatus.ROLLING_OUT)) + when(pipelineRepository.findByStatusIn(anyList())) .thenReturn(List.of(rollingOutPipeline())); when(applicationRuntimeGateway.getDeploymentHealth(any(), eq(NAMESPACE), eq(APP_NAME))) .thenReturn(new DeploymentHealth(false, false, 2, 1, null, Instant.now())); @@ -221,4 +233,81 @@ void pipelinesOfOneEnvironmentArePolledInASingleRequest() { verify(pipelineRepository).updateStatusAndMessageIfMatch( eq(second.getId()), eq(PipelineStatus.RUNNING), eq(PipelineStatus.ERROR), any()); } + + private static String lockOf(String pipelineId) { + return PipelineInstanceScanJob.PIPELINE_LOCK_PREFIX + pipelineId; + } + + /** A pipeline another server is driving is not even polled here; that server's scan owns it end to end. */ + @Test + void pipelineLockedByAnotherServerIsLeftToIt() { + when(pipelineRepository.findAllByStatus(PipelineStatus.RUNNING)).thenReturn(List.of(runningPipeline())); + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of(rollingOutPipeline())); + when(lockRegistry.tryAcquire(lockOf(RUNNING_PIPELINE_ID))).thenReturn(false); + when(lockRegistry.tryAcquire(lockOf(PIPELINE_ID))).thenReturn(false); + + scanJob.scanPipelineJobs(); + scanJob.scanRollingOutPipelines(); + + verify(pipelineJobGateway, never()).getStatuses(any(), any()); + verify(applicationRuntimeGateway, never()).getDeploymentHealth(any(), any(), any()); + verify(lockRegistry, never()).release(any()); + } + + @Test + void lockIsReleasedOnceThePipelineFinishes() { + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of(rollingOutPipeline())); + when(applicationRuntimeGateway.getDeploymentHealth(any(), eq(NAMESPACE), eq(APP_NAME))) + .thenReturn(new DeploymentHealth(false, true, 1, 1, null, null)); + when(pipelineRepository.updateStatusIfMatch(eq(PIPELINE_ID), eq(PipelineStatus.ROLLING_OUT), eq(PipelineStatus.SUCCEEDED))) + .thenReturn(1); + + scanJob.scanRollingOutPipelines(); + + verify(lockRegistry).tryAcquire(lockOf(PIPELINE_ID)); + verify(lockRegistry).release(lockOf(PIPELINE_ID)); + } + + @Test + void lockIsKeptWhileThePipelineIsStillRollingOut() { + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of(rollingOutPipeline())); + when(applicationRuntimeGateway.getDeploymentHealth(any(), eq(NAMESPACE), eq(APP_NAME))) + .thenReturn(new DeploymentHealth(false, false, 2, 1, null, Instant.now())); + + scanJob.scanRollingOutPipelines(); + + verify(lockRegistry, never()).release(any()); + } + + /** A failed build releases the pipeline too, whether the failure came from the Job or from the poll itself. */ + @Test + void lockIsReleasedWhenTheBuildFails() { + Pipeline pipeline = runningPipeline(); + when(pipelineRepository.findAllByStatus(PipelineStatus.RUNNING)).thenReturn(List.of(pipeline)); + when(pipelineJobGateway.getStatuses(any(), any())).thenReturn(Map.of(pipeline.getId(), PipelineJobStatus.FAILED)); + when(pipelineRepository.updateStatusAndMessageIfMatch(any(), eq(PipelineStatus.RUNNING), eq(PipelineStatus.ERROR), any())) + .thenReturn(1); + + scanJob.scanPipelineJobs(); + + verify(lockRegistry).release(lockOf(RUNNING_PIPELINE_ID)); + } + + /** + * A user can stop a pipeline from any server, and the stopped pipeline never comes back through a scan on the + * server that was driving it. Its lock is swept by comparing what was held before the scan with what is still + * active; a lock this scan takes itself is untouched, being newer than the snapshot. + */ + @Test + void lockOfPipelineFinishedElsewhereIsSwept() { + when(lockRegistry.heldLocks()).thenReturn(Set.of(lockOf("stopped-elsewhere"), lockOf(PIPELINE_ID))); + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of(rollingOutPipeline())); + when(applicationRuntimeGateway.getDeploymentHealth(any(), eq(NAMESPACE), eq(APP_NAME))) + .thenReturn(new DeploymentHealth(false, false, 2, 1, null, Instant.now())); + + scanJob.scanRollingOutPipelines(); + + verify(lockRegistry).release(lockOf("stopped-elsewhere")); + verify(lockRegistry, never()).release(lockOf(PIPELINE_ID)); + } } diff --git a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertOptInTests.java b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertOptInTests.java index 59044bba..1e7ef711 100644 --- a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertOptInTests.java +++ b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertOptInTests.java @@ -6,6 +6,7 @@ import com.github.wellch4n.oops.application.port.ResourceAlertProbe; import com.github.wellch4n.oops.application.service.EnvironmentService; import com.github.wellch4n.oops.infrastructure.config.ResourceAlertProperties; +import com.github.wellch4n.oops.infrastructure.lock.NamedLockRegistry; import com.github.wellch4n.oops.infrastructure.persistence.jpa.ApplicationAlertStateRepository; import com.github.wellch4n.oops.infrastructure.persistence.jpa.ApplicationRuntimeSpecRepository; import org.junit.jupiter.api.Test; @@ -24,7 +25,8 @@ class ResourceAlertOptInTests { .withBean(ApplicationRuntimeSpecRepository.class, () -> mock(ApplicationRuntimeSpecRepository.class)) .withBean(ApplicationAlertStateRepository.class, () -> mock(ApplicationAlertStateRepository.class)) .withBean(EnvironmentService.class, () -> mock(EnvironmentService.class)) - .withBean(ResourceAlertProbe.class, () -> mock(ResourceAlertProbe.class)); + .withBean(ResourceAlertProbe.class, () -> mock(ResourceAlertProbe.class)) + .withBean(NamedLockRegistry.class, () -> mock(NamedLockRegistry.class)); @Test void scanJobIsAbsentWhenNothingIsConfigured() { diff --git a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertScanJobTests.java b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertScanJobTests.java index 57a531ce..e2bbd52b 100644 --- a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertScanJobTests.java +++ b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ResourceAlertScanJobTests.java @@ -17,6 +17,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; @@ -67,6 +68,7 @@ public List podsSustainedAbove(Environment environment, String namespace private ApplicationAlertStateRepository alertStateRepository; private ResourceAlertProperties properties; private StubProbe probe; + private NamedLockRegistry lockRegistry; private List published; private ResourceAlertScanJob job; @@ -80,6 +82,8 @@ void setUp() { properties = new ResourceAlertProperties(); probe = new StubProbe(); published = new ArrayList<>(); + lockRegistry = mock(NamedLockRegistry.class); + when(lockRegistry.tryAcquire(ResourceAlertScanJob.LOCK_NAME)).thenReturn(true); Environment environment = new Environment(); environment.setName("prod"); @@ -88,7 +92,7 @@ void setUp() { when(runtimeSpecRepository.findAll()).thenReturn(List.of()); job = new ResourceAlertScanJob(runtimeSpecRepository, alertStateRepository, environmentService, - probe, properties, event -> { + probe, properties, lockRegistry, event -> { if (event instanceof ApplicationAlertEvent alertEvent) { published.add(alertEvent); } @@ -241,4 +245,18 @@ void sendsOneEventPerApplicationRatherThanPerPod() { assertEquals(3, published.getFirst().pods().size()); } + + /** Another server leads the scan: this one neither reads targets nor writes state, so nothing is notified twice. */ + @Test + void serverWithoutTheScanLockDoesNothing() { + when(lockRegistry.tryAcquire(ResourceAlertScanJob.LOCK_NAME)).thenReturn(false); + givenLimits(null, "2Gi"); + probe.firingPods.put(ResourceMetric.MEMORY, List.of("my-app-0")); + + job.scanResourceAlerts(); + + verify(runtimeSpecRepository, never()).findAll(); + verify(alertStateRepository, never()).saveAll(any()); + assertTrue(published.isEmpty()); + } } diff --git a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ScheduledRestartJobTests.java b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ScheduledRestartJobTests.java new file mode 100644 index 00000000..b666f9b0 --- /dev/null +++ b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/ScheduledRestartJobTests.java @@ -0,0 +1,79 @@ +package com.github.wellch4n.oops.infrastructure.scheduler; + +import static org.mockito.ArgumentMatchers.any; +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.port.ApplicationRuntimeGateway; +import com.github.wellch4n.oops.application.service.EnvironmentService; +import com.github.wellch4n.oops.domain.environment.Environment; +import com.github.wellch4n.oops.infrastructure.lock.NamedLockRegistry; +import com.github.wellch4n.oops.infrastructure.persistence.jpa.ApplicationExpertConfig; +import com.github.wellch4n.oops.infrastructure.persistence.jpa.ApplicationExpertConfigRepository; +import java.util.List; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * The restart scan is led by one server at a time: the one holding the job lock fires every due restart, the + * others do not even read the schedules. Two servers firing the same schedule would restart the application twice + * in the same minute. + */ +class ScheduledRestartJobTests { + + private ApplicationExpertConfigRepository expertConfigRepository; + private ApplicationRuntimeGateway applicationRuntimeGateway; + private NamedLockRegistry lockRegistry; + private Environment environment; + private ScheduledRestartJob job; + + @BeforeEach + void setUp() { + expertConfigRepository = mock(ApplicationExpertConfigRepository.class); + EnvironmentService environmentService = mock(EnvironmentService.class); + applicationRuntimeGateway = mock(ApplicationRuntimeGateway.class); + lockRegistry = mock(NamedLockRegistry.class); + + environment = new Environment(); + environment.setName("prod"); + when(environmentService.getEnvironments()).thenReturn(List.of(environment)); + when(expertConfigRepository.findAll()).thenReturn(List.of(restartEveryMinute())); + + job = new ScheduledRestartJob(expertConfigRepository, environmentService, applicationRuntimeGateway, lockRegistry); + } + + /** A schedule that is due on every tick, so the test does not depend on the wall clock. */ + private static ApplicationExpertConfig restartEveryMinute() { + ApplicationExpertConfig.EnvironmentConfig environmentConfig = new ApplicationExpertConfig.EnvironmentConfig(); + environmentConfig.setEnvironment("prod"); + environmentConfig.setScheduledRestartEnabled(true); + environmentConfig.setScheduledRestartCron("* * * * *"); + + ApplicationExpertConfig expertConfig = new ApplicationExpertConfig(); + expertConfig.setNamespace("ns"); + expertConfig.setApplicationName("app"); + expertConfig.setEnvironmentConfigs(List.of(environmentConfig)); + return expertConfig; + } + + @Test + void serverHoldingTheScanLockFiresDueRestarts() { + when(lockRegistry.tryAcquire(ScheduledRestartJob.LOCK_NAME)).thenReturn(true); + + job.scanScheduledRestarts(); + + verify(applicationRuntimeGateway).rolloutRestart(environment, "ns", "app"); + } + + @Test + void serverWithoutTheScanLockLeavesRestartsToTheLeader() { + when(lockRegistry.tryAcquire(ScheduledRestartJob.LOCK_NAME)).thenReturn(false); + + job.scanScheduledRestarts(); + + verify(expertConfigRepository, never()).findAll(); + verify(applicationRuntimeGateway, never()).rolloutRestart(any(), any(), any()); + } +} From f0ad9eb9fb7813043a6892cd13075d9a88af9d41 Mon Sep 17 00:00:00 2001 From: wellCh4n Date: Fri, 4 Sep 2026 19:05:15 +0800 Subject: [PATCH 2/2] fix(pipeline): make every status write conditional and recover stuck pipelines Four gaps either let a stale read overwrite a transition made elsewhere or left a pipeline in a state nothing would ever move it out of, which blocks the application from being deployed again: - stopPipeline saved the whole row unconditionally. With the scan job carrying a RUNNING pipeline into DEPLOYING on any server in between, the stop overwrote the deploy in progress, which then ran to completion under a pipeline that said it was stopped. It now transitions from the status the caller read and fails with "state changed concurrently" when that read is stale. - completeDeployPhase (scan job and service) ignored the result of the DEPLOYING -> ROLLING_OUT claim, so a pipeline stopped while its artifact was applied was announced as rolling out anyway. A lost claim is now logged and nothing further is reported; the deploy and rollback failure paths honour their claims the same way. - A RUNNING pipeline whose build Job the cluster no longer had (deleted by hand, work namespace cleaned, TTL-reaped) mapped to UNKNOWN and matched no branch, so it stayed RUNNING for good. A missing Job now fails the build at once; a Job that exists without a status is still left for the next tick. Stopping such a pipeline no longer errors on the missing Job either. - Nothing looked at DEPLOYING, so a server dying mid-deploy left the pipeline there forever. The rollout scan now fails a pipeline that has stayed DEPLOYING for ten minutes, measured from when the scanning server first saw it deploying, since no column records status-change time and the conditional updates bypass entity timestamps. Co-Authored-By: Claude Fable 5.1 --- AGENTS.md | 2 +- .../application/service/PipelineService.java | 63 ++++--- .../KubernetesPipelineJobGateway.java | 10 +- .../scheduler/PipelineInstanceScanJob.java | 97 ++++++++-- .../service/PipelineStopTests.java | 167 ++++++++++++++++++ .../PipelineVerificationScanTests.java | 134 +++++++++++++- 6 files changed, 430 insertions(+), 43 deletions(-) create mode 100644 src/test/java/com/github/wellch4n/oops/application/service/PipelineStopTests.java diff --git a/AGENTS.md b/AGENTS.md index 5898fa29..c78841a5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -173,7 +173,7 @@ 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: 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/scheduler/PipelineInstanceScanJob.java b/src/main/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineInstanceScanJob.java index 03909b15..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 @@ -24,6 +24,7 @@ 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; @@ -51,6 +52,7 @@ @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( @@ -58,6 +60,8 @@ public class PipelineInstanceScanJob { 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; @@ -128,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); @@ -176,17 +184,7 @@ 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); @@ -222,6 +220,30 @@ private void failPipeline(Pipeline pipeline, Exception exception) { 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 + )); + } + } + /** * Polls pipelines awaiting Kubernetes rollout. Each ROLLING_OUT pipeline is checked against the * live StatefulSet rollout: a converged rollout marks it SUCCEEDED, a missing workload, fatal pod state, or @@ -238,6 +260,10 @@ public void scanRollingOutPipelines() { 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; } @@ -267,11 +293,50 @@ public void scanRollingOutPipelines() { } } 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)); } @@ -332,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/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 events = ArgumentCaptor.forClass(Object.class); + verify(eventPublisher, Mockito.atLeast(0)).publishEvent(events.capture()); + return events.getAllValues().stream() + .filter(PipelineNotificationEvent.class::isInstance) + .map(event -> ((PipelineNotificationEvent) event).type()) + .toList(); + } + + @Test + void stopIsAConditionalTransitionFromTheStatusTheCallerSaw() { + pipelineIn(PipelineStatus.RUNNING); + when(pipelineRepository.updateStatusIfMatch(PIPELINE_ID, PipelineStatus.RUNNING, PipelineStatus.STOPPED)).thenReturn(1); + + assertTrue(pipelineService.stopPipeline(NAMESPACE, APP_NAME, PIPELINE_ID, OPERATOR)); + + verify(pipelineJobGateway).stop(any(), eq(APP_NAME + "-pipeline-" + PIPELINE_ID)); + verify(pipelineRepository, never()).save(any()); + assertTrue(publishedTypes().contains(PipelineNotificationType.STOPPED)); + } + + /** The scan job carried the pipeline into DEPLOYING between the read and the stop: the stop loses, loudly. */ + @Test + void stopLosesToAConcurrentTransition() { + pipelineIn(PipelineStatus.RUNNING); + when(pipelineRepository.updateStatusIfMatch(PIPELINE_ID, PipelineStatus.RUNNING, PipelineStatus.STOPPED)).thenReturn(0); + + assertThrows(BizException.class, () -> pipelineService.stopPipeline(NAMESPACE, APP_NAME, PIPELINE_ID, OPERATOR)); + + verify(pipelineRepository, never()).save(any()); + assertFalse(publishedTypes().contains(PipelineNotificationType.STOPPED)); + } + + /** A built pipeline has no Job left, so the cluster is not touched; the transition is still conditional. */ + @Test + void stoppingABuiltPipelineLeavesTheClusterAlone() { + pipelineIn(PipelineStatus.BUILD_SUCCEEDED); + when(pipelineRepository.updateStatusIfMatch(PIPELINE_ID, PipelineStatus.BUILD_SUCCEEDED, PipelineStatus.STOPPED)).thenReturn(1); + + assertTrue(pipelineService.stopPipeline(NAMESPACE, APP_NAME, PIPELINE_ID, OPERATOR)); + + verify(pipelineJobGateway, never()).stop(any(), any()); + verify(pipelineRepository).updateStatusIfMatch(PIPELINE_ID, PipelineStatus.BUILD_SUCCEEDED, PipelineStatus.STOPPED); + } + + /** + * A manual deploy whose pipeline was stopped while the artifact was being applied loses the DEPLOYING -> + * ROLLING_OUT claim and must not announce a rollout on top of the stop. + */ + @Test + void deployStoppedMidwayIsNotReportedAsRollingOut() { + pipelineIn(PipelineStatus.BUILD_SUCCEEDED); + when(pipelineRepository.updateStatusIfMatch(PIPELINE_ID, PipelineStatus.BUILD_SUCCEEDED, PipelineStatus.DEPLOYING)).thenReturn(1); + when(pipelineRepository.updateStatusIfMatch(PIPELINE_ID, PipelineStatus.DEPLOYING, PipelineStatus.ROLLING_OUT)).thenReturn(0); + + assertTrue(pipelineService.deployPipeline(NAMESPACE, APP_NAME, PIPELINE_ID, OPERATOR)); + + List published = publishedTypes(); + assertTrue(published.contains(PipelineNotificationType.DEPLOYING)); + assertFalse(published.contains(PipelineNotificationType.ROLLING_OUT)); + } +} diff --git a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineVerificationScanTests.java b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineVerificationScanTests.java index 07de3add..7edcbf14 100644 --- a/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineVerificationScanTests.java +++ b/src/test/java/com/github/wellch4n/oops/infrastructure/scheduler/PipelineVerificationScanTests.java @@ -1,5 +1,7 @@ package com.github.wellch4n.oops.infrastructure.scheduler; +import static org.junit.jupiter.api.Assertions.assertFalse; +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; @@ -9,6 +11,8 @@ import static org.mockito.Mockito.when; import com.github.wellch4n.oops.application.dto.DeploymentHealth; +import com.github.wellch4n.oops.application.event.PipelineNotificationEvent; +import com.github.wellch4n.oops.application.event.PipelineNotificationType; import com.github.wellch4n.oops.application.port.ApplicationRuntimeGateway; import com.github.wellch4n.oops.application.port.ArtifactDeploymentExecutor; import com.github.wellch4n.oops.application.port.PipelineJobGateway; @@ -16,6 +20,7 @@ import com.github.wellch4n.oops.application.port.repository.ApplicationRepository; import com.github.wellch4n.oops.application.port.repository.PipelineRepository; import com.github.wellch4n.oops.application.service.EnvironmentService; +import com.github.wellch4n.oops.domain.application.Application; import com.github.wellch4n.oops.domain.delivery.Pipeline; import com.github.wellch4n.oops.domain.delivery.PipelineStateMachine; import com.github.wellch4n.oops.domain.environment.Environment; @@ -23,12 +28,14 @@ import com.github.wellch4n.oops.infrastructure.lock.NamedLockRegistry; import com.github.wellch4n.oops.shared.exception.EnvironmentUnreachableException; import java.net.SocketTimeoutException; +import java.time.Duration; import java.time.Instant; import java.util.List; import java.util.Map; import java.util.Set; 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; @@ -49,6 +56,7 @@ class PipelineVerificationScanTests { private static final String RUNNING_PIPELINE_ID = "n6rmtcfmj3ongcvwkfirzls5"; private PipelineRepository pipelineRepository; + private ApplicationRepository applicationRepository; private EnvironmentService environmentService; private ApplicationRuntimeGateway applicationRuntimeGateway; private PipelineJobGateway pipelineJobGateway; @@ -61,7 +69,7 @@ void setUp() { pipelineRepository = Mockito.mock(PipelineRepository.class); environmentService = Mockito.mock(EnvironmentService.class); applicationRuntimeGateway = Mockito.mock(ApplicationRuntimeGateway.class); - ApplicationRepository applicationRepository = Mockito.mock(ApplicationRepository.class); + applicationRepository = Mockito.mock(ApplicationRepository.class); eventPublisher = Mockito.mock(ApplicationEventPublisher.class); pipelineJobGateway = Mockito.mock(PipelineJobGateway.class); ArtifactDeploymentExecutor artifactDeploymentExecutor = Mockito.mock(ArtifactDeploymentExecutor.class); @@ -310,4 +318,128 @@ void lockOfPipelineFinishedElsewhereIsSwept() { verify(lockRegistry).release(lockOf("stopped-elsewhere")); verify(lockRegistry, never()).release(lockOf(PIPELINE_ID)); } + + private Pipeline deployingPipeline() { + Pipeline pipeline = runningPipeline("deploying-id"); + pipeline.setStatus(PipelineStatus.DEPLOYING); + return pipeline; + } + + private List publishedTypes() { + ArgumentCaptor events = ArgumentCaptor.forClass(Object.class); + verify(eventPublisher, Mockito.atLeast(0)).publishEvent(events.capture()); + return events.getAllValues().stream() + .filter(PipelineNotificationEvent.class::isInstance) + .map(event -> ((PipelineNotificationEvent) event).type()) + .toList(); + } + + /** + * A Job the cluster no longer has (deleted by hand, work namespace cleaned up, or reaped by its TTL) is never + * coming back; the pipeline is failed rather than left RUNNING for good, which would block the application. + */ + @Test + void missingBuildJobFailsThePipeline() { + Pipeline pipeline = runningPipeline(); + when(pipelineRepository.findAllByStatus(PipelineStatus.RUNNING)).thenReturn(List.of(pipeline)); + when(pipelineJobGateway.getStatuses(any(), any())).thenReturn(Map.of()); + when(pipelineRepository.updateStatusAndMessageIfMatch(eq(RUNNING_PIPELINE_ID), eq(PipelineStatus.RUNNING), eq(PipelineStatus.ERROR), any())) + .thenReturn(1); + + scanJob.scanPipelineJobs(); + + verify(pipelineRepository).updateStatusAndMessageIfMatch( + eq(RUNNING_PIPELINE_ID), eq(PipelineStatus.RUNNING), eq(PipelineStatus.ERROR), any()); + assertTrue(publishedTypes().contains(PipelineNotificationType.FAILED)); + verify(lockRegistry).release(lockOf(RUNNING_PIPELINE_ID)); + } + + /** A Job that exists but carries no status yet is not a missing Job; it is simply not finished. */ + @Test + void buildJobWithoutStatusYetIsLeftAlone() { + Pipeline pipeline = runningPipeline(); + when(pipelineRepository.findAllByStatus(PipelineStatus.RUNNING)).thenReturn(List.of(pipeline)); + when(pipelineJobGateway.getStatuses(any(), any())).thenReturn(Map.of(pipeline.getId(), PipelineJobStatus.UNKNOWN)); + + scanJob.scanPipelineJobs(); + + verify(pipelineRepository, never()).updateStatusAndMessageIfMatch(any(), any(), any(), any()); + verify(pipelineRepository, never()).updateStatusIfMatch(any(), any(), any()); + } + + /** + * Stopped from another server while the artifact was being applied: the DEPLOYING -> ROLLING_OUT claim loses, + * and the pipeline must not be announced as rolling out on top of having been stopped. + */ + @Test + void pipelineStoppedWhileDeployingIsNotReportedAsRollingOut() { + Pipeline pipeline = runningPipeline(); + Application application = new Application(); + application.setName(APP_NAME); + application.setNamespace(NAMESPACE); + when(pipelineRepository.findAllByStatus(PipelineStatus.RUNNING)).thenReturn(List.of(pipeline)); + when(pipelineJobGateway.getStatuses(any(), any())).thenReturn(Map.of(pipeline.getId(), PipelineJobStatus.SUCCEEDED)); + when(pipelineRepository.updateStatusIfMatch(RUNNING_PIPELINE_ID, PipelineStatus.RUNNING, PipelineStatus.DEPLOYING)).thenReturn(1); + when(applicationRepository.findAggregate(NAMESPACE, APP_NAME)).thenReturn(application); + when(pipelineRepository.updateStatusIfMatch(RUNNING_PIPELINE_ID, PipelineStatus.DEPLOYING, PipelineStatus.ROLLING_OUT)).thenReturn(0); + + scanJob.scanPipelineJobs(); + + List published = publishedTypes(); + assertTrue(published.contains(PipelineNotificationType.DEPLOYING)); + assertFalse(published.contains(PipelineNotificationType.ROLLING_OUT)); + } + + /** The first sight of a deploying pipeline only starts its clock; a deploy normally finishes within seconds. */ + @Test + void freshDeployIsLeftAlone() { + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of(deployingPipeline())); + + scanJob.scanRollingOutPipelines(); + + verify(pipelineRepository, never()).updateStatusAndMessageIfMatch(any(), any(), any(), any()); + assertTrue(scanJob.deployingSince.containsKey("deploying-id")); + } + + /** + * A pipeline still DEPLOYING long after this server first saw it was abandoned by whoever claimed it — a server + * that died mid-deploy — and is the only thing standing between the application and its next deploy. + */ + @Test + void deployStuckBeyondTimeoutIsFailed() { + Pipeline pipeline = deployingPipeline(); + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of(pipeline)); + when(pipelineRepository.updateStatusAndMessageIfMatch(eq("deploying-id"), eq(PipelineStatus.DEPLOYING), eq(PipelineStatus.ERROR), any())) + .thenReturn(1); + scanJob.deployingSince.put("deploying-id", Instant.now().minus(Duration.ofMinutes(11))); + + scanJob.scanRollingOutPipelines(); + + verify(pipelineRepository).updateStatusAndMessageIfMatch( + eq("deploying-id"), eq(PipelineStatus.DEPLOYING), eq(PipelineStatus.ERROR), any()); + assertTrue(publishedTypes().contains(PipelineNotificationType.FAILED)); + verify(lockRegistry).release(lockOf("deploying-id")); + } + + /** A stuck deploy another server is still driving (it holds the lock) is that server's to time out. */ + @Test + void stuckDeployLockedElsewhereIsLeftToItsServer() { + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of(deployingPipeline())); + when(lockRegistry.tryAcquire(lockOf("deploying-id"))).thenReturn(false); + scanJob.deployingSince.put("deploying-id", Instant.now().minus(Duration.ofMinutes(11))); + + scanJob.scanRollingOutPipelines(); + + verify(pipelineRepository, never()).updateStatusAndMessageIfMatch(any(), any(), any(), any()); + } + + @Test + void deployThatMovedOnIsForgotten() { + scanJob.deployingSince.put("deploying-id", Instant.now()); + when(pipelineRepository.findByStatusIn(anyList())).thenReturn(List.of()); + + scanJob.scanRollingOutPipelines(); + + assertFalse(scanJob.deployingSince.containsKey("deploying-id")); + } }