entries = new ConcurrentHashMap<>();
+ private final LongSupplier nanoTime;
+
+ public NextCheckSchedule() {
+ this(System::nanoTime);
+ }
+
+ /** 시계를 주입하는 생성자 — 간격은 시간에 의존하므로 테스트가 시계를 쥘 수 있어야 한다. */
+ public NextCheckSchedule(LongSupplier nanoTime) {
+ this.nanoTime = nanoTime;
+ }
+
+ /** 지금 이 대상을 볼 차례인가. 장부에 없으면(처음 보는 대상) 언제나 그렇다. */
+ public boolean due(K key) {
+ Entry entry = entries.get(key);
+ return entry == null || nanoTime.getAsLong() - entry.nextCheckAtNanos >= 0;
+ }
+
+ /** 이 대상을 몇 번이나 결론 없이 봤는가 — 호출자가 간격 단계를 고르는 데 쓴다. */
+ public int inconclusiveChecks(K key) {
+ Entry entry = entries.get(key);
+ return entry == null ? 0 : entry.inconclusiveChecks;
+ }
+
+ /** 이 대상을 {@code delayMillis} 뒤에 다시 본다. */
+ public void scheduleAfter(K key, long delayMillis) {
+ long now = nanoTime.getAsLong();
+ entries.compute(key, (ignored, previous) -> new Entry(
+ now + TimeUnit.MILLISECONDS.toNanos(delayMillis),
+ previous == null ? 1 : previous.inconclusiveChecks + 1,
+ now
+ ));
+ evictStale(now);
+ }
+
+ /** 결론이 났다 — 장부에서 지운다. 같은 대상이 다시 나타나면 처음부터 센다. */
+ public void clear(K key) {
+ entries.remove(key);
+ }
+
+ private void evictStale(long now) {
+ long ttlNanos = TimeUnit.MILLISECONDS.toNanos(ENTRY_TTL_MS);
+ entries.values().removeIf(entry -> now - entry.touchedAtNanos > ttlNanos);
+ }
+
+ private record Entry(long nextCheckAtNanos, int inconclusiveChecks, long touchedAtNanos) {
+ }
+}
diff --git a/src/main/java/com/example/dvely/common/worker/WorkQueue.java b/src/main/java/com/example/dvely/common/worker/WorkQueue.java
new file mode 100644
index 00000000..4eaec212
--- /dev/null
+++ b/src/main/java/com/example/dvely/common/worker/WorkQueue.java
@@ -0,0 +1,20 @@
+package com.example.dvely.common.worker;
+
+/**
+ * 1초 폴링 워커가 감시하는 작업 큐. {@link WorkerPollGate} 의 백오프 상태와 깨우기 신호가
+ * 이 값 단위로 갈린다 — 배포 큐에 일이 들어왔다고 웹훅 워커까지 깨울 이유는 없다.
+ */
+public enum WorkQueue {
+
+ /** {@code agent_runs} — 사용자 메시지로 만들어진 Agent 태스크. */
+ AGENT_RUN,
+
+ /** {@code deployment_histories} — 사용자가 누른 배포. */
+ DEPLOYMENT_RUN,
+
+ /** {@code cloud_connection_verification_jobs} — 사용자가 요청한 클라우드 권한 검증. */
+ CLOUD_CONNECTION_VERIFICATION,
+
+ /** {@code webhook_deliveries} — GitHub 이 보내온 웹훅. */
+ WEBHOOK_DELIVERY
+}
diff --git a/src/main/java/com/example/dvely/common/worker/WorkQueuedEvent.java b/src/main/java/com/example/dvely/common/worker/WorkQueuedEvent.java
new file mode 100644
index 00000000..d30a64c9
--- /dev/null
+++ b/src/main/java/com/example/dvely/common/worker/WorkQueuedEvent.java
@@ -0,0 +1,12 @@
+package com.example.dvely.common.worker;
+
+/**
+ * 어떤 큐에 워커가 집을 수 있는 일이 들어갔다는 신호(#340 5-1).
+ *
+ * 커밋된 뒤에 전달돼야 한다 — 커밋 전에 깨우면 워커가 아직 보이지 않는 행을 찾다가
+ * 헛폴링하고 도로 백오프에 들어간다. 그래서 {@link WorkerPollGate} 의 수신부가
+ * {@code @TransactionalEventListener(AFTER_COMMIT)} 이다. 발행부는 그냥 {@code publishEvent}
+ * 하면 되고, 트랜잭션 안이면 커밋 뒤로 미뤄지고 밖이면 즉시 전달된다.
+ */
+public record WorkQueuedEvent(WorkQueue queue) {
+}
diff --git a/src/main/java/com/example/dvely/common/worker/WorkerPollGate.java b/src/main/java/com/example/dvely/common/worker/WorkerPollGate.java
new file mode 100644
index 00000000..ac98ed37
--- /dev/null
+++ b/src/main/java/com/example/dvely/common/worker/WorkerPollGate.java
@@ -0,0 +1,145 @@
+package com.example.dvely.common.worker;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+/**
+ * 1초 폴링 워커들의 적응형 백오프 게이트(#340 5-1).
+ *
+ * 왜 필요한가. 유휴(사용자 0명·작업 0건) 상태에서 워커 4종이 매초 DB 를 쳐서 분당
+ * 1904 쿼리가 나갔다. 그 비용의 대부분은 SELECT 자체가 아니라 트랜잭션 의례다 — 폴링 한 번이
+ * {@code SET autocommit=0} → SELECT → {@code COMMIT} → {@code SET autocommit=1} 로 왕복 4배가
+ * 된다. 일이 없을 때 폴링 횟수를 줄이면 의례까지 같은 비율로 줄어든다.
+ *
+ * 왜 {@code @Scheduled} 간격을 직접 늘리지 않는가. 워커의 {@code fixedDelay} 는 1초로
+ * 두고 이 게이트가 DB 를 칠지 말지만 정한다. 스케줄러 스레드가 1초마다 깨어나는 비용은
+ * 인메모리 비교 한 번이고, 대신 두 가지를 공짜로 얻는다 — (1) 깨우기 신호가 오면 다음 틱(≤1초)에
+ * 곧바로 폴링하므로 첫 claim 지연이 백오프와 무관하게 항상 1초 이내다. 동적
+ * {@code Trigger} 로 간격 자체를 늘렸다면 깨우기가 재스케줄까지 해야 했다. (2) 스케줄러
+ * 인프라를 손대지 않으므로 워커별로 켜고 끄기가 자유롭다.
+ *
+ * 깨우기와 폴백 폴링은 둘 다 필요하다. {@link WorkQueuedEvent} 는 이 JVM 안에서만
+ * 도는 인스턴스 로컬 신호다. 다중 인스턴스로 늘리면 다른 인스턴스가 넣은 일은 이 신호를
+ * 타고 오지 않는다. 그래서 백오프에는 반드시 상한이 있고(기본 30초), 상한에 닿아도 폴링은
+ * 멈추지 않는다 — 최악의 경우에도 남의 일을 30초 안에는 본다.
+ *
+ * 깨우기 유실 방지. 폴링이 도는 도중에 신호가 오면, 그 신호가 가리키는 행을 방금 끝난
+ * 폴링이 못 봤을 수 있다. 그때 백오프를 늘리면 이미 들어온 일이 최대 상한만큼 늦어진다. 그래서
+ * 폴링 시작 시점의 세대를 기억해 두고, 끝났을 때 세대가 달라져 있으면 백오프를 늘리지 않는다.
+ */
+@Component
+public class WorkerPollGate {
+
+ private final long baseDelayMs;
+ private final long maxDelayMs;
+ private final LongSupplier nanoTime;
+ private final Map states = new EnumMap<>(WorkQueue.class);
+
+ @Autowired
+ public WorkerPollGate(@Value("${qeploy.worker.poll.base-delay-ms:1000}") long baseDelayMs,
+ @Value("${qeploy.worker.poll.max-delay-ms:30000}") long maxDelayMs) {
+ this(baseDelayMs, maxDelayMs, System::nanoTime);
+ }
+
+ /** 시계를 주입하는 생성자 — 백오프는 시간에 의존하므로 테스트가 시계를 직접 쥘 수 있어야 한다. */
+ public WorkerPollGate(long baseDelayMs, long maxDelayMs, LongSupplier nanoTime) {
+ this.baseDelayMs = Math.max(1L, baseDelayMs);
+ this.maxDelayMs = Math.max(this.baseDelayMs, maxDelayMs);
+ this.nanoTime = nanoTime;
+ long now = nanoTime.getAsLong();
+ for (WorkQueue queue : WorkQueue.values()) {
+ states.put(queue, new QueueState(this.baseDelayMs, now));
+ }
+ }
+
+ /** 이번 틱에 DB 를 쳐도 되는가. {@code false} 면 워커는 아무 쿼리도 내지 않고 돌아간다. */
+ public boolean shouldPoll(WorkQueue queue) {
+ return states.get(queue).beginPoll(nanoTime.getAsLong());
+ }
+
+ /**
+ * 이번 폴링이 일을 찾았다(또는 실행기 포화로 claim 을 미뤘다). 간격을 최소로 되돌린다 —
+ * 일이 있는 동안에는 빠르게 도는 것이 맞다.
+ */
+ public void recordBusy(WorkQueue queue) {
+ states.get(queue).recordBusy(nanoTime.getAsLong());
+ }
+
+ /** 이번 폴링이 빈손이었다. 다음 간격을 두 배로 늘린다(상한까지). */
+ public void recordIdle(WorkQueue queue) {
+ states.get(queue).recordIdle(nanoTime.getAsLong(), maxDelayMs);
+ }
+
+ /** 백오프를 즉시 풀어 다음 틱에 폴링하게 한다. */
+ public void wake(WorkQueue queue) {
+ states.get(queue).wake(nanoTime.getAsLong());
+ }
+
+ /**
+ * enqueue 가 커밋된 뒤 해당 큐의 워커를 깨운다.
+ *
+ * {@code fallbackExecution = true} 인 이유: 발행부가 항상 트랜잭션 안이라고 가정할 수
+ * 없다. 트랜잭션 밖에서 발행된 신호를 조용히 버리면, 그 큐는 백오프 상한만큼 늦게 깨어난다 —
+ * 사용자에게는 "눌렀는데 한참 아무 일도 안 일어남"으로 보인다. 밖이면 즉시 처리한다.
+ */
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
+ public void onWorkQueued(WorkQueuedEvent event) {
+ wake(event.queue());
+ }
+
+ /**
+ * 큐 하나의 백오프 상태. 갱신은 폴링 스레드(스케줄러)와 깨우기 스레드(요청·실행기) 양쪽에서
+ * 오지만 초당 몇 번 수준이라, 눈에 보이게 맞는 {@code synchronized} 를 쓴다 — 여기서 lock-free
+ * 로 얻을 것은 없고, 잃을 것(깨우기 유실)은 사용자에게 최대 30초로 보인다.
+ */
+ private static final class QueueState {
+
+ private final long baseDelayMs;
+ private long delayMs;
+ private long nextPollAtNanos;
+ private long generation;
+ private long polledGeneration;
+
+ private QueueState(long baseDelayMs, long now) {
+ this.baseDelayMs = baseDelayMs;
+ this.delayMs = baseDelayMs;
+ this.nextPollAtNanos = now; // 기동 직후 첫 틱은 무조건 폴링한다
+ }
+
+ private synchronized boolean beginPoll(long now) {
+ if (now - nextPollAtNanos < 0) {
+ return false;
+ }
+ polledGeneration = generation;
+ return true;
+ }
+
+ private synchronized void recordBusy(long now) {
+ delayMs = baseDelayMs;
+ nextPollAtNanos = now;
+ }
+
+ private synchronized void recordIdle(long now, long max) {
+ if (generation != polledGeneration) {
+ // 폴링이 도는 중에 깨우기가 왔다. 그 일을 이 폴링이 못 봤을 수 있으므로 백오프를
+ // 늘리지 않는다 — wake 가 이미 간격과 다음 시각을 최소로 돌려놨다.
+ return;
+ }
+ delayMs = Math.min(delayMs * 2, max);
+ nextPollAtNanos = now + TimeUnit.MILLISECONDS.toNanos(delayMs);
+ }
+
+ private synchronized void wake(long now) {
+ generation++;
+ delayMs = baseDelayMs;
+ nextPollAtNanos = now;
+ }
+ }
+}
diff --git a/src/main/java/com/example/dvely/deployment/application/command/DeploymentCommandService.java b/src/main/java/com/example/dvely/deployment/application/command/DeploymentCommandService.java
index 79499e7a..74d5f371 100644
--- a/src/main/java/com/example/dvely/deployment/application/command/DeploymentCommandService.java
+++ b/src/main/java/com/example/dvely/deployment/application/command/DeploymentCommandService.java
@@ -23,6 +23,7 @@
import com.example.dvely.deployment.application.service.DeploymentOutcomeService;
import com.example.dvely.deployment.domain.model.DeploymentHistory;
import com.example.dvely.deployment.domain.repository.DeploymentHistoryRepository;
+import com.example.dvely.deployment.infrastructure.worker.DeploymentExecutionRegistry;
import com.example.dvely.deployment.domain.value.DeployTargetType;
import com.example.dvely.deployment.domain.value.PackageManager;
import com.example.dvely.deployment.infrastructure.workflow.DeployWorkflowTemplate;
@@ -64,6 +65,9 @@ public class DeploymentCommandService {
private final GithubActionsPort githubActionsPort;
private final GithubRepoPort githubRepoPort;
private final DeploymentHistoryRepository deploymentHistoryRepository;
+ // #340 5-8: executeQueued 가 끝났음을 하트비트에 알리는 유일한 지점. 등록은 DeploymentRunWorker
+ // 가 제출 직전에 한다(그쪽 javadoc 참고) — 여기서 등록하면 executor 큐 대기 창이 비어버린다.
+ private final DeploymentExecutionRegistry deploymentExecutionRegistry;
// Track Z (#56) D1/§5.4: needed so a direct deploy can no longer silently merge preview into
// main once the result-approval gate owns that project (see prepareRelease's mergeAllowed).
private final ProjectApprovalPolicyRepository projectApprovalPolicyRepository;
@@ -233,6 +237,10 @@ public void executeQueued(Long historyId) {
execute(historyId);
} catch (Exception exception) {
handleExecutionFailure(historyId, exception);
+ } finally {
+ // #340 5-8: 이 배포는 더 이상 이 인스턴스가 돌리고 있지 않다 — 하트비트 대상에서
+ // 빼야 한다. 여기서 빼지 않으면 이미 끝난 이력의 리스를 30초마다 되살리게 된다.
+ deploymentExecutionRegistry.unregister(historyId);
}
}
diff --git a/src/main/java/com/example/dvely/deployment/domain/repository/DeploymentHistoryRepository.java b/src/main/java/com/example/dvely/deployment/domain/repository/DeploymentHistoryRepository.java
index 4d1eeee8..ba372c96 100644
--- a/src/main/java/com/example/dvely/deployment/domain/repository/DeploymentHistoryRepository.java
+++ b/src/main/java/com/example/dvely/deployment/domain/repository/DeploymentHistoryRepository.java
@@ -3,6 +3,7 @@
import com.example.dvely.deployment.domain.model.DeploymentHistory;
import com.example.dvely.project.domain.value.DeployStatus;
import java.time.LocalDateTime;
+import java.util.Collection;
import java.util.List;
import java.util.Optional;
@@ -33,7 +34,32 @@ public interface DeploymentHistoryRepository {
List claimPending(String workerId, int limit);
+ /**
+ * 폴링 한 번이 하는 일 전부 — 만료 리스 회수와 claim 을 한 트랜잭션으로 묶는다(#340 5-1).
+ * 따로 부르면 폴링 한 번이 트랜잭션 두 개가 되고, 트랜잭션마다 붙는 {@code SET autocommit} ·
+ * {@code COMMIT} 의례가 유휴 DB 비용의 대부분이었다. 회수 UPDATE 가 같은 트랜잭션에서 먼저
+ * 반영되므로, 방금 회수된 행을 같은 폴링의 claim 이 곧바로 집는다.
+ */
+ List recoverAndClaimPending(String workerId, int limit);
+
void recoverExpiredLeases();
- void renewLeases(String workerId);
+ /**
+ * claim 해 놓고 실행기에 넘기지 못한 이력을 PENDING 으로 되돌린다(#340 5-2).
+ *
+ * 이 호출이 없으면 배치의 두 번째 이력이 IN_PROGRESS 인 채 리스 만료(2분)까지 방치된다 —
+ * 사용자 화면에는 "배포 중"으로 보이지만 그것을 실제로 돌리는 스레드는 어디에도 없다.
+ *
+ * @return 이 호출이 실제로 되돌렸으면 true. false 는 그 사이 다른 주체(리스 만료 회수 등)가
+ * 이미 이 행을 IN_PROGRESS 밖으로 옮겼다는 뜻이고, 그것도 정상이다.
+ */
+ boolean releaseClaim(Long historyId, String workerId, long backoffMillis);
+
+ /**
+ * 이 인스턴스가 실제로 실행 중인 이력들의 리스만 연장한다(#340 5-8).
+ *
+ * {@code historyIds} 는 호출자의 실행 레지스트리 스냅샷이다. 비어 있으면 호출자가 아예
+ * 부르지 않는다 — 실행 중인 배포가 없는데 0행짜리 UPDATE 를 30초마다 내보낼 이유가 없다.
+ */
+ void renewLeases(String workerId, Collection historyIds);
}
diff --git a/src/main/java/com/example/dvely/deployment/infrastructure/persistence/repository/DeploymentHistoryRepositoryAdapter.java b/src/main/java/com/example/dvely/deployment/infrastructure/persistence/repository/DeploymentHistoryRepositoryAdapter.java
index e007bbc3..1e39f284 100644
--- a/src/main/java/com/example/dvely/deployment/infrastructure/persistence/repository/DeploymentHistoryRepositoryAdapter.java
+++ b/src/main/java/com/example/dvely/deployment/infrastructure/persistence/repository/DeploymentHistoryRepositoryAdapter.java
@@ -2,13 +2,17 @@
import com.example.dvely.deployment.domain.model.DeploymentHistory;
import com.example.dvely.deployment.domain.repository.DeploymentHistoryRepository;
+import com.example.dvely.common.worker.WorkQueue;
+import com.example.dvely.common.worker.WorkQueuedEvent;
import com.example.dvely.deployment.infrastructure.persistence.entity.DeploymentHistoryEntity;
import com.example.dvely.project.domain.value.DeployStatus;
+import java.util.Collection;
import java.util.List;
import java.util.Optional;
import java.time.Duration;
import java.time.LocalDateTime;
import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
import org.springframework.data.domain.PageRequest;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;
@@ -18,9 +22,21 @@
public class DeploymentHistoryRepositoryAdapter implements DeploymentHistoryRepository {
private final SpringDataDeploymentHistoryRepository springDataRepository;
+ private final ApplicationEventPublisher eventPublisher;
@Override
public DeploymentHistory save(DeploymentHistory history) {
+ DeploymentHistory saved = doSave(history);
+ if (saved.getStatus() == DeployStatus.PENDING) {
+ // #340 5-1: 워커가 집을 수 있는 상태로 저장됐다 — 커밋 뒤에 워커를 깨운다. 이 신호가
+ // 없으면 유휴가 길어져 폴링 간격이 상한까지 늘어난 뒤 배포 버튼을 누른 사용자가 그
+ // 상한만큼 기다린다.
+ eventPublisher.publishEvent(new WorkQueuedEvent(WorkQueue.DEPLOYMENT_RUN));
+ }
+ return saved;
+ }
+
+ private DeploymentHistory doSave(DeploymentHistory history) {
if (history.getId() == null) {
DeploymentHistoryEntity entity = springDataRepository.save(DeploymentHistoryEntity.from(history));
return entity.toDomain();
@@ -99,6 +115,13 @@ public List claimPending(String workerId, int limit) {
.toList();
}
+ @Override
+ @Transactional
+ public List recoverAndClaimPending(String workerId, int limit) {
+ recoverExpiredLeases();
+ return claimPending(workerId, limit);
+ }
+
@Override
@Transactional
public void recoverExpiredLeases() {
@@ -118,11 +141,27 @@ public void recoverExpiredLeases() {
@Override
@Transactional
- public void renewLeases(String workerId) {
+ public boolean releaseClaim(Long historyId, String workerId, long backoffMillis) {
+ return springDataRepository.releaseClaim(
+ historyId,
+ workerId,
+ LocalDateTime.now().plus(Duration.ofMillis(backoffMillis)),
+ DeployStatus.IN_PROGRESS.name(),
+ DeployStatus.PENDING.name()
+ ) == 1;
+ }
+
+ @Override
+ @Transactional
+ public void renewLeases(String workerId, Collection historyIds) {
+ if (historyIds.isEmpty()) {
+ return;
+ }
springDataRepository.renewLeases(
workerId,
LocalDateTime.now().plusMinutes(2),
- DeployStatus.IN_PROGRESS.name()
+ DeployStatus.IN_PROGRESS.name(),
+ historyIds
);
}
}
diff --git a/src/main/java/com/example/dvely/deployment/infrastructure/persistence/repository/SpringDataDeploymentHistoryRepository.java b/src/main/java/com/example/dvely/deployment/infrastructure/persistence/repository/SpringDataDeploymentHistoryRepository.java
index e47b346c..6689576c 100644
--- a/src/main/java/com/example/dvely/deployment/infrastructure/persistence/repository/SpringDataDeploymentHistoryRepository.java
+++ b/src/main/java/com/example/dvely/deployment/infrastructure/persistence/repository/SpringDataDeploymentHistoryRepository.java
@@ -2,6 +2,7 @@
import com.example.dvely.deployment.infrastructure.persistence.entity.DeploymentHistoryEntity;
import java.time.LocalDateTime;
+import java.util.Collection;
import java.util.List;
import java.util.Optional;
import org.springframework.data.domain.Pageable;
@@ -54,21 +55,52 @@ int claim(
@Param("runningStatus") String runningStatus
);
+ // #340 5-2: claim 된 이력을 실행기에 넘기지 못했을 때 PENDING 으로 되돌린다. claim 이 올린
+ // attempt 를 그대로 되돌리는 이유는, 실행기 포화가 이 배포의 실패가 아니라 시스템 부하로 인한
+ // 배압이기 때문이다 — 재시도 예산을 여기서 쓰면 사용자는 아무 잘못 없이 재시도 횟수를 잃는다
+ // (AgentRunWorker 의 ADR-Y3 releaseClaim 이 attempt 를 건드리지 않는 것과 같은 판단).
+ // WHERE 는 claim 과 같은 조건부 UPDATE 모양(id + status + lease_owner)이라, 이 claim 을 실제로
+ // 쥔 워커만 되돌릴 수 있고 동시에 도는 recoverExpiredLeases 와 이중 전이를 만들지 않는다.
+ @Modifying(clearAutomatically = true, flushAutomatically = true)
+ @Query("""
+ update DeploymentHistoryEntity history
+ set history.status = :pendingStatus,
+ history.attempt = history.attempt - 1,
+ history.nextRunAt = :nextRunAt,
+ history.leaseOwner = null,
+ history.leaseUntil = null
+ where history.id = :historyId
+ and history.status = :runningStatus
+ and history.leaseOwner = :workerId
+ """)
+ int releaseClaim(
+ @Param("historyId") Long historyId,
+ @Param("workerId") String workerId,
+ @Param("nextRunAt") LocalDateTime nextRunAt,
+ @Param("runningStatus") String runningStatus,
+ @Param("pendingStatus") String pendingStatus
+ );
+
List findByStatusAndLeaseUntilBefore(String status, LocalDateTime now);
List findByStatusAndLeaseUntilIsNullAndUpdatedAtBefore(
String status, LocalDateTime updatedBefore, Pageable pageable);
+ // #340 5-8: 갱신 범위를 호출자의 DeploymentExecutionRegistry 스냅샷(historyIds)으로 좁힌다.
+ // 이 워커가 claim 했지만 실행기가 거부한 이력의 리스를 여기서 계속 살려두면
+ // recoverExpiredLeases 가 그것을 영영 회수하지 못한다(AgentRunWorker 의 ADR-Y4 와 같은 이유).
@Modifying(clearAutomatically = true, flushAutomatically = true)
@Query("""
update DeploymentHistoryEntity history
set history.leaseUntil = :leaseUntil
where history.status = :runningStatus
and history.leaseOwner = :workerId
+ and history.id in :historyIds
""")
int renewLeases(
@Param("workerId") String workerId,
@Param("leaseUntil") LocalDateTime leaseUntil,
- @Param("runningStatus") String runningStatus
+ @Param("runningStatus") String runningStatus,
+ @Param("historyIds") Collection historyIds
);
}
diff --git a/src/main/java/com/example/dvely/deployment/infrastructure/worker/DeploymentExecutionRegistry.java b/src/main/java/com/example/dvely/deployment/infrastructure/worker/DeploymentExecutionRegistry.java
new file mode 100644
index 00000000..d5b38889
--- /dev/null
+++ b/src/main/java/com/example/dvely/deployment/infrastructure/worker/DeploymentExecutionRegistry.java
@@ -0,0 +1,37 @@
+package com.example.dvely.deployment.infrastructure.worker;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import org.springframework.stereotype.Component;
+
+/**
+ * 이 인스턴스가 실제로 실행하기로 한 배포 이력 ID 의 JVM 로컬 레지스트리(#340 5-8).
+ *
+ * {@code DeploymentRunWorker#renewLeases} 의 하트비트가 이 집합으로 범위를 좁힌다. 예전에는
+ * "이 workerId 가 소유한 IN_PROGRESS 행 전부"를 조건 없이 갱신했고, 그래서 두 가지가 함께
+ * 나빴다 — (1) 실행 중인 배포가 하나도 없어도 30초마다 0행짜리 UPDATE 가 나갔고, (2) 실행기에
+ * 넘기지 못해 실행 주체가 없는 행의 리스까지 계속 살려두면 {@code recoverExpiredLeases} 가
+ * 그것을 영영 회수하지 못한다. {@code AgentRunWorker} 가 #55(ADR-Y4)에서 고친 것과 같은 형태다.
+ *
+ * 등록은 executor 제출 직전에 해야 한다. 제출은 됐지만 아직 executor 큐에서 대기 중인
+ * 배포도 하트비트 보호를 받아야 하기 때문이다 — 그 창을 비워두면 리스가 만료돼 회수가 같은
+ * 일을 다시 집는데, 원래 제출분은 나중에 그대로 실행된다(이중 실행).
+ */
+@Component
+public class DeploymentExecutionRegistry {
+
+ private final Set registeredHistoryIds = ConcurrentHashMap.newKeySet();
+
+ public void register(Long historyId) {
+ registeredHistoryIds.add(historyId);
+ }
+
+ public void unregister(Long historyId) {
+ registeredHistoryIds.remove(historyId);
+ }
+
+ /** 하트비트의 {@code history_id IN (...)} 필터에 쓸 시점 스냅샷. */
+ public Set snapshot() {
+ return Set.copyOf(registeredHistoryIds);
+ }
+}
diff --git a/src/main/java/com/example/dvely/deployment/infrastructure/worker/DeploymentRunWorker.java b/src/main/java/com/example/dvely/deployment/infrastructure/worker/DeploymentRunWorker.java
index 5b907b75..a3671f8c 100644
--- a/src/main/java/com/example/dvely/deployment/infrastructure/worker/DeploymentRunWorker.java
+++ b/src/main/java/com/example/dvely/deployment/infrastructure/worker/DeploymentRunWorker.java
@@ -1,35 +1,110 @@
package com.example.dvely.deployment.infrastructure.worker;
import com.example.dvely.deployment.application.command.DeploymentCommandService;
+import com.example.dvely.common.worker.WorkQueue;
+import com.example.dvely.common.worker.WorkerPollGate;
import com.example.dvely.deployment.domain.repository.DeploymentHistoryRepository;
import java.lang.management.ManagementFactory;
-import lombok.RequiredArgsConstructor;
+import java.util.List;
+import java.util.Set;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
+/**
+ * PENDING 배포를 집어 {@link DeploymentCommandService#executeQueued}(비동기) 로 넘긴다.
+ *
+ * #340 5-2: 위임을 이력 하나씩 격리한다. 예전에는 루프가 통째로 노출돼 있어,
+ * {@code deploymentExecutor} 가 포화된 순간 {@code @Async} 프록시가 던지는
+ * {@code TaskRejectedException} 하나가 루프를 끊었다. 그러면 같은 배치에서 이미 claim 된
+ * 다음 이력이 IN_PROGRESS 인 채로 남고, 그것을 돌리는 스레드는 어디에도 없으며, 리스 만료(2분)
+ * 전까지는 아무도 회수하지 않는다. 사용자 화면에는 그 2분 동안 "배포 중"이 떠 있다.
+ * {@code AgentRunWorker} 가 #55(ADR-Y3)에서 이미 고친 것과 같은 버그가 여기에만 남아 있었다.
+ */
@Slf4j
@Component
-@RequiredArgsConstructor
public class DeploymentRunWorker {
private static final int CLAIM_BATCH_SIZE = 2;
private final DeploymentHistoryRepository deploymentHistoryRepository;
private final DeploymentCommandService deploymentCommandService;
+ private final WorkerPollGate pollGate;
+ private final DeploymentExecutionRegistry executionRegistry;
+ private final long dispatchRejectBackoffMs;
private final String workerId = ManagementFactory.getRuntimeMXBean().getName() + "-deployment";
+ public DeploymentRunWorker(DeploymentHistoryRepository deploymentHistoryRepository,
+ DeploymentCommandService deploymentCommandService,
+ WorkerPollGate pollGate,
+ DeploymentExecutionRegistry executionRegistry,
+ @Value("${qeploy.deployment.worker.dispatch-reject-backoff-ms:5000}")
+ long dispatchRejectBackoffMs) {
+ this.deploymentHistoryRepository = deploymentHistoryRepository;
+ this.deploymentCommandService = deploymentCommandService;
+ this.pollGate = pollGate;
+ this.executionRegistry = executionRegistry;
+ this.dispatchRejectBackoffMs = dispatchRejectBackoffMs;
+ }
+
@Scheduled(fixedDelayString = "${qeploy.deployment.worker.poll-interval-ms:1000}")
public void dispatchPendingDeployments() {
- deploymentHistoryRepository.recoverExpiredLeases();
- for (Long historyId : deploymentHistoryRepository.claimPending(workerId, CLAIM_BATCH_SIZE)) {
+ // #340 5-1: 틱은 1초마다 오지만, 일이 없는 동안에는 DB 를 치지 않는다.
+ if (!pollGate.shouldPoll(WorkQueue.DEPLOYMENT_RUN)) {
+ return;
+ }
+ List historyIds;
+ try {
+ historyIds = deploymentHistoryRepository.recoverAndClaimPending(workerId, CLAIM_BATCH_SIZE);
+ } catch (RuntimeException exception) {
+ // DB 가 흔들리는 동안 매초 같은 쿼리를 던져봐야 소용이 없다 — 물러나며 재시도한다.
+ pollGate.recordIdle(WorkQueue.DEPLOYMENT_RUN);
+ throw exception;
+ }
+ if (historyIds.isEmpty()) {
+ pollGate.recordIdle(WorkQueue.DEPLOYMENT_RUN);
+ } else {
+ pollGate.recordBusy(WorkQueue.DEPLOYMENT_RUN);
+ }
+ for (Long historyId : historyIds) {
+ dispatchOne(historyId);
+ }
+ }
+
+ private void dispatchOne(Long historyId) {
+ // #340 5-8: 제출 "직전"에 등록한다. 제출은 됐지만 아직 executor 큐에서 대기 중인 배포도
+ // 하트비트 보호를 받아야 한다 — 그 창을 비워두면 리스가 만료돼 회수가 같은 일을 다시
+ // 집는데 원래 제출분은 나중에 그대로 실행된다.
+ executionRegistry.register(historyId);
+ try {
log.info("배포 Job 실행 위임: historyId={} workerId={}", historyId, workerId);
deploymentCommandService.executeQueued(historyId);
+ } catch (RuntimeException exception) {
+ // 실행기 포화(TaskRejectedException)든 제출 전 다른 예외든 결론은 같다 — 이 이력은
+ // 돌지 않는다. 그렇다면 claim 을 붙들고 있을 이유가 없으므로 즉시 PENDING 으로
+ // 되돌린다. 예외를 밖으로 흘리지 않는 것이 핵심이다: 흘리면 이 배치의 나머지가
+ // claim 된 채 버려진다.
+ handleDispatchFailure(historyId, exception);
}
}
+ private void handleDispatchFailure(Long historyId, RuntimeException exception) {
+ executionRegistry.unregister(historyId);
+ boolean released = deploymentHistoryRepository.releaseClaim(
+ historyId, workerId, dispatchRejectBackoffMs);
+ log.warn("배포 Job 위임 실패 — 재대기열로 반환합니다. historyId={} workerId={} released={} 원인={}",
+ historyId, workerId, released, exception.toString());
+ }
+
@Scheduled(fixedDelayString = "${qeploy.deployment.worker.heartbeat-interval-ms:30000}")
public void renewLeases() {
- deploymentHistoryRepository.renewLeases(workerId);
+ Set registered = executionRegistry.snapshot();
+ if (registered.isEmpty()) {
+ // #340 5-8: 이 JVM 이 지금 실제로 돌리고 있는 배포가 없다 — 쿼리를 아예 내지 않는다.
+ // (AgentRunWorker 는 이미 이렇게 하고 있었고, 이쪽만 조건 없이 UPDATE 를 내보냈다.)
+ return;
+ }
+ deploymentHistoryRepository.renewLeases(workerId, registered);
}
}
diff --git a/src/main/java/com/example/dvely/deployment/infrastructure/worker/StuckDeploymentRecoveryWorker.java b/src/main/java/com/example/dvely/deployment/infrastructure/worker/StuckDeploymentRecoveryWorker.java
index f8e9172a..cabe490d 100644
--- a/src/main/java/com/example/dvely/deployment/infrastructure/worker/StuckDeploymentRecoveryWorker.java
+++ b/src/main/java/com/example/dvely/deployment/infrastructure/worker/StuckDeploymentRecoveryWorker.java
@@ -1,6 +1,7 @@
package com.example.dvely.deployment.infrastructure.worker;
import com.example.dvely.auth.application.command.AuthCommandService;
+import com.example.dvely.common.worker.NextCheckSchedule;
import com.example.dvely.auth.domain.model.User;
import com.example.dvely.auth.domain.repository.UserRepository;
import com.example.dvely.deployment.application.port.out.GithubActionsPort;
@@ -12,6 +13,7 @@
import com.example.dvely.deployment.infrastructure.workflow.DeployWorkflowTemplate;
import com.example.dvely.project.domain.model.Project;
import com.example.dvely.project.domain.repository.ProjectRepository;
+import java.time.Duration;
import java.time.LocalDateTime;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -44,16 +46,35 @@ public class StuckDeploymentRecoveryWorker {
private final DeploymentOutcomeService deploymentOutcomeService;
private final StuckDeploymentRecoveryProperties properties;
+ /**
+ * 판정을 못 얻은 이력별 재조회 간격(#340 5-4). 예전에는 멈춘 배포 한 건이 포기 시각(기본
+ * 120분)에 닿을 때까지 매분 GitHub 을 쳤다 — 한 건당 최대 120회다. runId 가 없는
+ * 이력은 그때마다 목록 조회(per_page=30)까지 돌았다.
+ *
+ * GitHub Actions 실행 상태는 초 단위로 바뀌지 않는다. 첫 몇 번은 촘촘히 보되(방금 끝났을
+ * 수 있다) 계속 결론이 없으면 물러난다 — 1 → 2 → 5 → 10분. 판정이 나면 장부에서 지운다.
+ */
+ private final NextCheckSchedule recheckSchedule = new NextCheckSchedule<>();
+
+ /** 결론 없는 재조회의 간격 단계(분). 마지막 값에 닿으면 그 값을 유지한다. */
+ private static final long[] RECHECK_BACKOFF_MINUTES = {1L, 2L, 5L, 10L};
+
@Scheduled(fixedDelayString = "${qeploy.deployment.recovery.poll-interval-ms:60000}")
public void recoverStuckDeployments() {
LocalDateTime graceCutoff = LocalDateTime.now().minusMinutes(properties.graceMinutesOrDefault());
for (DeploymentHistory history : deploymentHistoryRepository.findDispatchedAwaitingOutcome(
graceCutoff, properties.batchSizeOrDefault())) {
+ if (!recheckSchedule.due(history.getId())) {
+ continue; // #340 5-4: 아직 다시 물을 때가 아니다 — GitHub 호출을 아끼는 유일한 지점
+ }
try {
recover(history);
} catch (RuntimeException exception) {
// 한 이력의 실패가 나머지를 막지 않는다. GitHub 호출은 레이트 리밋·토큰 만료로
- // 언제든 실패할 수 있고 다음 주기에 다시 물으면 되는 성격이다.
+ // 언제든 실패할 수 있고 다음 주기에 다시 물으면 되는 성격이다. 실패도 "판정을 못
+ // 얻은 것"이므로 같은 백오프를 태운다 — 레이트 리밋에 걸린 상태로 매분 다시 치는
+ // 것이 가장 나쁘다.
+ backOffRecheck(history.getId());
log.warn("멈춘 배포 회수 실패 — 다음 주기에 재시도: historyId={} 원인={}",
history.getId(), exception.toString());
}
@@ -65,6 +86,8 @@ private void recover(DeploymentHistory history) {
if (project == null) {
log.warn("멈춘 배포의 프로젝트가 없음: historyId={} projectId={}",
history.getId(), history.getProjectId());
+ // 프로젝트가 없으면 다음 주기에도 없다 — 매분 다시 확인할 이유가 없다.
+ backOffRecheck(history.getId());
return;
}
@@ -80,12 +103,24 @@ private void recover(DeploymentHistory history) {
if ("success".equals(run.conclusion())) {
log.info("멈춘 배포 회수 — 성공으로 확정: historyId={} runId={}", history.getId(), run.runId());
deploymentOutcomeService.applySuccess(history, project);
+ recheckSchedule.clear(history.getId());
return;
}
log.info("멈춘 배포 회수 — 실패로 확정: historyId={} runId={} conclusion={}",
history.getId(), run.runId(), run.conclusion());
deploymentOutcomeService.applyFailure(history, project,
DeployFailureCode.WORKFLOW_FAILED, "GitHub Actions workflow conclusion: " + run.conclusion());
+ recheckSchedule.clear(history.getId());
+ }
+
+ /**
+ * 이 이력을 다음에 언제 다시 물을지 미룬다(#340 5-4). 결론 없는 조회가 이어질수록 간격이
+ * 1 → 2 → 5 → 10분으로 늘어난다.
+ */
+ private void backOffRecheck(Long historyId) {
+ int checks = recheckSchedule.inconclusiveChecks(historyId);
+ long minutes = RECHECK_BACKOFF_MINUTES[Math.min(checks, RECHECK_BACKOFF_MINUTES.length - 1)];
+ recheckSchedule.scheduleAfter(historyId, Duration.ofMinutes(minutes).toMillis());
}
/**
@@ -124,12 +159,15 @@ private GithubActionsPort.WorkflowRunStatus readRunStatus(DeploymentHistory hist
private void abandonIfHopeless(DeploymentHistory history, Project project, String reason) {
LocalDateTime abandonCutoff = LocalDateTime.now().minusMinutes(properties.abandonMinutesOrDefault());
if (history.getUpdatedAt() == null || history.getUpdatedAt().isAfter(abandonCutoff)) {
+ // 아직 포기할 때가 아니다 = 이번 조회는 판정을 못 얻었다. 다음 조회를 미룬다.
+ backOffRecheck(history.getId());
return;
}
log.warn("멈춘 배포 회수 포기 — 결과 미확인으로 닫는다: historyId={} runId={} 사유={}",
history.getId(), history.getWorkflowRunId(), reason);
deploymentOutcomeService.applyFailure(history, project,
DeployFailureCode.RESULT_UNKNOWN, "배포 결과를 확인할 수 없습니다. " + reason);
+ recheckSchedule.clear(history.getId());
}
private String resolveUserToken(Long ownerUserId) {
diff --git a/src/main/java/com/example/dvely/domainbinding/infrastructure/worker/DomainVerificationWorker.java b/src/main/java/com/example/dvely/domainbinding/infrastructure/worker/DomainVerificationWorker.java
index d35f2f44..7cb81256 100644
--- a/src/main/java/com/example/dvely/domainbinding/infrastructure/worker/DomainVerificationWorker.java
+++ b/src/main/java/com/example/dvely/domainbinding/infrastructure/worker/DomainVerificationWorker.java
@@ -1,14 +1,17 @@
package com.example.dvely.domainbinding.infrastructure.worker;
+import com.example.dvely.common.worker.NextCheckSchedule;
import com.example.dvely.domainbinding.application.command.DomainBindingCommandService;
import com.example.dvely.domainbinding.domain.model.DomainBinding;
import com.example.dvely.domainbinding.domain.repository.DomainBindingRepository;
import com.example.dvely.domainbinding.domain.value.DomainStatus;
import com.example.dvely.domainbinding.domain.value.DomainType;
import com.example.dvely.domainbinding.infrastructure.config.DomainVerificationProperties;
+import java.time.Duration;
import java.time.LocalDateTime;
-import lombok.RequiredArgsConstructor;
+import java.util.concurrent.Executor;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@@ -27,7 +30,6 @@
*/
@Slf4j
@Component
-@RequiredArgsConstructor
public class DomainVerificationWorker {
// CONNECTED 후 HTTPS 뱃지(httpsEnforced)를 채우려 재검증하는 창. EC2(Caddy on-demand TLS)는 첫 https
@@ -35,9 +37,42 @@ public class DomainVerificationWorker {
// 보내 warming)으로 곧 true 가 된다. 무한 프로브를 막으려 생성 후 이 창 안에서만 재검증한다(#6).
private static final int HTTPS_RECHECK_WINDOW_MINUTES = 30;
+ // #340 5-5: 생성 후 경과 시간별 재검증 간격. 갓 만든 도메인은 곧 붙을 수 있으니 촘촘히 보고,
+ // 한 시간이 지나도 안 붙었으면 사용자가 registrar 설정을 아직 안 했을 가능성이 높다 —
+ // 그때부터는 성기게 본다.
+ private static final Duration YOUNG_DOMAIN_WINDOW = Duration.ofMinutes(10);
+ private static final Duration YOUNG_DOMAIN_INTERVAL = Duration.ofMinutes(1);
+ private static final Duration MIDDLE_AGED_DOMAIN_WINDOW = Duration.ofHours(1);
+ private static final Duration MIDDLE_AGED_DOMAIN_INTERVAL = Duration.ofMinutes(5);
+ private static final Duration OLD_DOMAIN_INTERVAL = Duration.ofMinutes(10);
+
private final DomainBindingRepository domainBindingRepository;
private final DomainBindingCommandService domainBindingCommandService;
private final DomainVerificationProperties properties;
+ private final Executor verificationExecutor;
+
+ /**
+ * 도메인별 재검증 간격(#340 5-5). 예전에는 배치 전체를 매 주기(60초) 검증했다 —
+ * 커스텀 도메인은 TTL 이 1440분이므로 한 건당 최대 1,440회, 그것도 Cloudflare + GitHub +
+ * HTTPS 프로브 세 묶음이다.
+ *
+ * 기다리는 대상이 DNS 전파와 호스팅 반영이라 시간이 지날수록 "다음 1분 안에 바뀔"
+ * 확률이 떨어진다. 그래서 갓 만든 도메인은 촘촘히, 오래된 도메인은 성기게 본다.
+ */
+ private final NextCheckSchedule verifySchedule = new NextCheckSchedule<>();
+
+ /** CONNECTED 후 HTTPS 뱃지 재검증도 같은 이유로 간격을 둔다 — 위와 키가 겹치면 안 되므로 따로 둔다. */
+ private final NextCheckSchedule httpsRecheckSchedule = new NextCheckSchedule<>();
+
+ public DomainVerificationWorker(DomainBindingRepository domainBindingRepository,
+ DomainBindingCommandService domainBindingCommandService,
+ DomainVerificationProperties properties,
+ @Qualifier("domainVerificationExecutor") Executor verificationExecutor) {
+ this.domainBindingRepository = domainBindingRepository;
+ this.domainBindingCommandService = domainBindingCommandService;
+ this.properties = properties;
+ this.verificationExecutor = verificationExecutor;
+ }
@Scheduled(fixedDelayString = "${qeploy.domain-binding.verification.poll-interval-ms:60000}")
public void verifyPendingDomains() {
@@ -49,8 +84,13 @@ public void verifyPendingDomains() {
}
if (isExpired(domain)) {
abandon(domain);
+ verifySchedule.clear(domain.getId());
continue;
}
+ if (!verifySchedule.due(domain.getId())) {
+ continue; // #340 5-5: 아직 다시 볼 때가 아니다 — 외부 API 호출을 아끼는 지점
+ }
+ verifySchedule.scheduleAfter(domain.getId(), recheckIntervalFor(domain).toMillis());
verify(domain);
}
recheckHttpsForConnectedDomains();
@@ -71,13 +111,19 @@ private void recheckHttpsForConnectedDomains() {
if (!withinHttpsRecheckWindow(domain)) {
continue; // 창을 지났는데도 https 미확인 — 수동 재검증에 맡긴다(무한 프로브 방지).
}
- try {
- // verify() 와 달리 "검증 완료" 로그를 남기지 않는다 — 이미 CONNECTED 라 매 주기 스팸이 된다.
- domainBindingCommandService.checkVerificationAsSystem(domain.getId());
- } catch (RuntimeException exception) {
- log.debug("HTTPS 재검증 실패(다음 주기 재시도): domainId={} 원인={}",
- domain.getId(), exception.toString());
+ if (!httpsRecheckSchedule.due(domain.getId())) {
+ continue;
}
+ httpsRecheckSchedule.scheduleAfter(domain.getId(), recheckIntervalFor(domain).toMillis());
+ submit(() -> {
+ try {
+ // verify() 와 달리 "검증 완료" 로그를 남기지 않는다 — 이미 CONNECTED 라 매 주기 스팸이 된다.
+ domainBindingCommandService.checkVerificationAsSystem(domain.getId());
+ } catch (RuntimeException exception) {
+ log.debug("HTTPS 재검증 실패(다음 주기 재시도): domainId={} 원인={}",
+ domain.getId(), exception.toString());
+ }
+ }, domain);
}
}
@@ -87,20 +133,59 @@ private boolean withinHttpsRecheckWindow(DomainBinding domain) {
&& createdAt.plusMinutes(HTTPS_RECHECK_WINDOW_MINUTES).isAfter(LocalDateTime.now());
}
+ /**
+ * #340 5-5: 검증 자체를 전용 executor 로 넘긴다. Cloudflare + GitHub + HTTPS 프로브 세 묶음이
+ * 배치 20건만큼 직렬로 스케줄러 스레드를 붙들고 있었고, 한 건이 타임아웃까지 버티면 그
+ * 시간이 그대로 스케줄러 점유가 됐다 — 같은 풀을 쓰는 다른 잡까지 함께 밀린다.
+ */
private void verify(DomainBinding domain) {
- try {
- // 한 도메인의 실패가 나머지를 막지 않아야 한다. Cloudflare·GitHub 호출이 섞여 있어
- // 어느 하나는 언제든 실패할 수 있고, 다음 주기에 다시 시도하면 되는 성격이다.
- DomainStatus status = domainBindingCommandService
- .checkVerificationAsSystem(domain.getId())
- .status();
- if (status == DomainStatus.CONNECTED) {
- log.info("도메인 검증 완료: domainId={} hostname={}", domain.getId(), domain.getHostname());
+ submit(() -> {
+ try {
+ // 한 도메인의 실패가 나머지를 막지 않아야 한다. Cloudflare·GitHub 호출이 섞여 있어
+ // 어느 하나는 언제든 실패할 수 있고, 다음 주기에 다시 시도하면 되는 성격이다.
+ DomainStatus status = domainBindingCommandService
+ .checkVerificationAsSystem(domain.getId())
+ .status();
+ if (status == DomainStatus.CONNECTED) {
+ log.info("도메인 검증 완료: domainId={} hostname={}", domain.getId(), domain.getHostname());
+ verifySchedule.clear(domain.getId());
+ }
+ } catch (RuntimeException exception) {
+ log.warn("도메인 검증 실패 — 다음 주기에 재시도: domainId={} hostname={} 원인={}",
+ domain.getId(), domain.getHostname(), exception.toString());
}
+ }, domain);
+ }
+
+ /**
+ * executor 제출 자체가 실패해도(포화) 이 도메인의 이번 차례만 건너뛴다. 검증에는 claim 이
+ * 없어 되돌릴 상태가 없다 — 다음 차례에 다시 집으면 그만이다. 다만 간격은 이미 잡혔으므로
+ * "바로 다음 주기"가 아니라 그 간격 뒤에 다시 온다는 점을 알고 넘긴다: 포화는 우리가 이미
+ * 많이 돌고 있다는 뜻이라, 그 상황에서 서둘러 재시도하는 것이 옳지 않다.
+ */
+ private void submit(Runnable task, DomainBinding domain) {
+ try {
+ verificationExecutor.execute(task);
} catch (RuntimeException exception) {
- log.warn("도메인 검증 실패 — 다음 주기에 재시도: domainId={} hostname={} 원인={}",
- domain.getId(), domain.getHostname(), exception.toString());
+ log.warn("도메인 검증 위임 실패 — 다음 차례에 재시도: domainId={} 원인={}",
+ domain.getId(), exception.toString());
+ }
+ }
+
+ /** 생성 후 경과 시간이 길수록 성기게 본다. createdAt 이 없으면 가장 촘촘한 간격을 쓴다. */
+ private Duration recheckIntervalFor(DomainBinding domain) {
+ LocalDateTime createdAt = domain.getCreatedAt();
+ if (createdAt == null) {
+ return YOUNG_DOMAIN_INTERVAL;
+ }
+ Duration age = Duration.between(createdAt, LocalDateTime.now());
+ if (age.compareTo(YOUNG_DOMAIN_WINDOW) <= 0) {
+ return YOUNG_DOMAIN_INTERVAL;
+ }
+ if (age.compareTo(MIDDLE_AGED_DOMAIN_WINDOW) <= 0) {
+ return MIDDLE_AGED_DOMAIN_INTERVAL;
}
+ return OLD_DOMAIN_INTERVAL;
}
private void abandon(DomainBinding domain) {
diff --git a/src/main/java/com/example/dvely/webhook/domain/repository/WebhookDeliveryRepository.java b/src/main/java/com/example/dvely/webhook/domain/repository/WebhookDeliveryRepository.java
index ed1077db..16c13b0d 100644
--- a/src/main/java/com/example/dvely/webhook/domain/repository/WebhookDeliveryRepository.java
+++ b/src/main/java/com/example/dvely/webhook/domain/repository/WebhookDeliveryRepository.java
@@ -15,6 +15,14 @@ public interface WebhookDeliveryRepository {
List claimPending(String workerId, int limit);
+ /**
+ * 폴링 한 번이 하는 일 전부 — 만료 리스 회수와 claim 을 한 트랜잭션으로 묶는다(#340 5-1).
+ * 따로 부르면 폴링 한 번이 트랜잭션 두 개가 되고, 트랜잭션마다 붙는 {@code SET autocommit} ·
+ * {@code COMMIT} 의례가 유휴 DB 비용의 대부분이었다. 회수 UPDATE 가 같은 트랜잭션에서 먼저
+ * 반영되므로, 방금 회수된 행을 같은 폴링의 claim 이 곧바로 집는다.
+ */
+ List recoverAndClaimPending(String workerId, int limit);
+
void recoverExpiredLeases();
/**
@@ -31,4 +39,11 @@ public interface WebhookDeliveryRepository {
* 호출자는 0 이 돌아올 때까지 반복한다.
*/
int deleteTerminalBatch(LocalDateTime cutoff, int batchSize);
+
+ /**
+ * claim 해 놓고 executor 에 넘기지 못한 배달을 PENDING 으로 되돌린다(#340 5-3).
+ *
+ * @return 이 호출이 실제로 되돌렸으면 true.
+ */
+ boolean releaseClaim(String deliveryId, String workerId, long backoffMillis);
}
diff --git a/src/main/java/com/example/dvely/webhook/infrastructure/persistence/repository/SpringDataWebhookDeliveryRepository.java b/src/main/java/com/example/dvely/webhook/infrastructure/persistence/repository/SpringDataWebhookDeliveryRepository.java
index aacf97f1..dc759d33 100644
--- a/src/main/java/com/example/dvely/webhook/infrastructure/persistence/repository/SpringDataWebhookDeliveryRepository.java
+++ b/src/main/java/com/example/dvely/webhook/infrastructure/persistence/repository/SpringDataWebhookDeliveryRepository.java
@@ -45,6 +45,30 @@ int claim(
@Param("processingStatus") String processingStatus
);
+ // #340 5-3: claim 한 배달을 executor 에 넘기지 못했을 때 되돌린다. claim 이 올린 attempt 를
+ // 그대로 되돌리는 이유는 실행기 포화가 이 배달의 실패가 아니기 때문이다 — 재시도 예산을
+ // 여기서 쓰면 GitHub 이 다시 보내주지 않는 배달을 우리가 스스로 버리게 된다. WHERE 는 claim 과
+ // 같은 조건부 UPDATE 모양이라 이 claim 을 실제로 쥔 워커만 되돌린다.
+ @Modifying(clearAutomatically = true, flushAutomatically = true)
+ @Query("""
+ update WebhookDeliveryEntity delivery
+ set delivery.status = :pendingStatus,
+ delivery.attempt = delivery.attempt - 1,
+ delivery.nextAttemptAt = :nextAttemptAt,
+ delivery.leaseOwner = null,
+ delivery.leaseUntil = null
+ where delivery.id = :deliveryId
+ and delivery.status = :processingStatus
+ and delivery.leaseOwner = :workerId
+ """)
+ int releaseClaim(
+ @Param("deliveryId") String deliveryId,
+ @Param("workerId") String workerId,
+ @Param("nextAttemptAt") LocalDateTime nextAttemptAt,
+ @Param("processingStatus") String processingStatus,
+ @Param("pendingStatus") String pendingStatus
+ );
+
List findByStatusAndLeaseUntilBefore(String status, LocalDateTime now);
/**
diff --git a/src/main/java/com/example/dvely/webhook/infrastructure/persistence/repository/WebhookDeliveryRepositoryAdapter.java b/src/main/java/com/example/dvely/webhook/infrastructure/persistence/repository/WebhookDeliveryRepositoryAdapter.java
index fd68d894..f45ebb22 100644
--- a/src/main/java/com/example/dvely/webhook/infrastructure/persistence/repository/WebhookDeliveryRepositoryAdapter.java
+++ b/src/main/java/com/example/dvely/webhook/infrastructure/persistence/repository/WebhookDeliveryRepositoryAdapter.java
@@ -1,13 +1,17 @@
package com.example.dvely.webhook.infrastructure.persistence.repository;
+import com.example.dvely.common.worker.WorkQueue;
+import com.example.dvely.common.worker.WorkQueuedEvent;
import com.example.dvely.webhook.domain.model.WebhookDelivery;
import com.example.dvely.webhook.domain.repository.WebhookDeliveryRepository;
import com.example.dvely.webhook.domain.value.WebhookDeliveryStatus;
import com.example.dvely.webhook.infrastructure.persistence.entity.WebhookDeliveryEntity;
+import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Optional;
import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.data.domain.PageRequest;
import org.springframework.stereotype.Repository;
@@ -38,6 +42,7 @@ public class WebhookDeliveryRepositoryAdapter implements WebhookDeliveryReposito
);
private final SpringDataWebhookDeliveryRepository springDataRepository;
+ private final ApplicationEventPublisher eventPublisher;
@Override
public boolean enqueue(WebhookDelivery delivery) {
@@ -46,6 +51,9 @@ public boolean enqueue(WebhookDelivery delivery) {
}
try {
springDataRepository.saveAndFlush(WebhookDeliveryEntity.from(delivery));
+ // #340 5-1: 새 배달이 큐에 들어갔다 — 커밋 뒤에 워커를 깨운다. GitHub 은 응답이 늦으면
+ // 재전송하므로, 유휴 백오프가 늘어난 상태에서 배달이 상한만큼 방치되면 안 된다.
+ eventPublisher.publishEvent(new WorkQueuedEvent(WorkQueue.WEBHOOK_DELIVERY));
return true;
} catch (DataIntegrityViolationException exception) {
return false;
@@ -87,6 +95,25 @@ public List claimPending(String workerId, int limit) {
.toList();
}
+ @Override
+ @Transactional
+ public boolean releaseClaim(String deliveryId, String workerId, long backoffMillis) {
+ return springDataRepository.releaseClaim(
+ deliveryId,
+ workerId,
+ LocalDateTime.now().plus(Duration.ofMillis(backoffMillis)),
+ WebhookDeliveryStatus.PROCESSING.name(),
+ WebhookDeliveryStatus.PENDING.name()
+ ) == 1;
+ }
+
+ @Override
+ @Transactional
+ public List recoverAndClaimPending(String workerId, int limit) {
+ recoverExpiredLeases();
+ return claimPending(workerId, limit);
+ }
+
@Override
@Transactional
public void recoverExpiredLeases() {
diff --git a/src/main/java/com/example/dvely/webhook/infrastructure/worker/WebhookDeliveryWorker.java b/src/main/java/com/example/dvely/webhook/infrastructure/worker/WebhookDeliveryWorker.java
index 0cfe2e2d..1afc1e2a 100644
--- a/src/main/java/com/example/dvely/webhook/infrastructure/worker/WebhookDeliveryWorker.java
+++ b/src/main/java/com/example/dvely/webhook/infrastructure/worker/WebhookDeliveryWorker.java
@@ -1,30 +1,105 @@
package com.example.dvely.webhook.infrastructure.worker;
+import com.example.dvely.common.worker.WorkQueue;
+import com.example.dvely.common.worker.WorkerPollGate;
import com.example.dvely.webhook.application.WebhookService;
import com.example.dvely.webhook.domain.repository.WebhookDeliveryRepository;
import java.lang.management.ManagementFactory;
-import lombok.RequiredArgsConstructor;
+import java.util.List;
+import java.util.concurrent.Executor;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
+/**
+ * PENDING·RETRY_WAIT 웹훅 배달을 집어 전용 executor 로 넘긴다.
+ *
+ * #340 5-3: 예전에는 배달 처리를 스케줄러 스레드에서 동기로 돌렸다. 핸들러가 GitHub API
+ * 를 호출하므로 한 배달이 느리면 그 동안 이 워커의 다음 폴링이 통째로 밀리고, 스케줄러 풀을
+ * 공유하는 다른 잡까지 함께 굶었다. 한 폴링이 최대 10건을 직렬로 처리했으므로 최악은 그 10배다.
+ *
+ * 비동기로 넘기면 {@code TaskRejectedException} 이라는 새 실패 경로가 생긴다 — 그래서 배달
+ * 하나씩 격리하고, 거부되면 claim 을 즉시 PENDING 으로 되돌린다({@code DeploymentRunWorker} ·
+ * {@code AgentRunWorker} 와 같은 ADR-Y3 패턴). 되돌리지 않으면 그 배달은 리스 만료(2분)까지
+ * PROCESSING 인 채 남고, GitHub 은 우리가 200 을 이미 준 배달을 다시 보내주지 않는다.
+ */
@Slf4j
@Component
-@RequiredArgsConstructor
public class WebhookDeliveryWorker {
private static final int CLAIM_BATCH_SIZE = 10;
private final WebhookDeliveryRepository webhookDeliveryRepository;
private final WebhookService webhookService;
+ private final WorkerPollGate pollGate;
+ private final Executor webhookExecutor;
+ private final long dispatchRejectBackoffMs;
private final String workerId = ManagementFactory.getRuntimeMXBean().getName() + "-webhook";
+ public WebhookDeliveryWorker(WebhookDeliveryRepository webhookDeliveryRepository,
+ WebhookService webhookService,
+ WorkerPollGate pollGate,
+ @Qualifier("webhookExecutor") Executor webhookExecutor,
+ @Value("${qeploy.webhook.worker.dispatch-reject-backoff-ms:5000}")
+ long dispatchRejectBackoffMs) {
+ this.webhookDeliveryRepository = webhookDeliveryRepository;
+ this.webhookService = webhookService;
+ this.pollGate = pollGate;
+ this.webhookExecutor = webhookExecutor;
+ this.dispatchRejectBackoffMs = dispatchRejectBackoffMs;
+ }
+
@Scheduled(fixedDelayString = "${qeploy.webhook.worker.poll-interval-ms:1000}")
public void dispatchPendingDeliveries() {
- webhookDeliveryRepository.recoverExpiredLeases();
- for (String deliveryId : webhookDeliveryRepository.claimPending(workerId, CLAIM_BATCH_SIZE)) {
- log.info("webhook delivery 처리: deliveryId={} workerId={}", deliveryId, workerId);
+ // #340 5-1: 틱은 1초마다 오지만, 일이 없는 동안에는 DB 를 치지 않는다. 새 배달이 들어오면
+ // enqueue 커밋 뒤에 오는 깨우기 신호가 백오프를 즉시 풀어, 다음 틱(≤1초)에 집힌다.
+ if (!pollGate.shouldPoll(WorkQueue.WEBHOOK_DELIVERY)) {
+ return;
+ }
+ List deliveryIds;
+ try {
+ deliveryIds = webhookDeliveryRepository.recoverAndClaimPending(workerId, CLAIM_BATCH_SIZE);
+ } catch (RuntimeException exception) {
+ // DB 가 흔들리는 동안 매초 같은 쿼리를 던져봐야 소용이 없다 — 물러나며 재시도한다.
+ pollGate.recordIdle(WorkQueue.WEBHOOK_DELIVERY);
+ throw exception;
+ }
+ if (deliveryIds.isEmpty()) {
+ pollGate.recordIdle(WorkQueue.WEBHOOK_DELIVERY);
+ } else {
+ pollGate.recordBusy(WorkQueue.WEBHOOK_DELIVERY);
+ }
+ for (String deliveryId : deliveryIds) {
+ dispatchOne(deliveryId);
+ }
+ }
+
+ private void dispatchOne(String deliveryId) {
+ try {
+ log.info("webhook delivery 처리 위임: deliveryId={} workerId={}", deliveryId, workerId);
+ webhookExecutor.execute(() -> processSafely(deliveryId));
+ } catch (RuntimeException exception) { // TaskRejectedException(실행기 포화) 포함
+ boolean released = webhookDeliveryRepository.releaseClaim(
+ deliveryId, workerId, dispatchRejectBackoffMs);
+ log.warn("webhook delivery 위임 실패 — 재대기열로 반환합니다. deliveryId={} workerId={} released={} 원인={}",
+ deliveryId, workerId, released, exception.toString());
+ }
+ }
+
+ /**
+ * executor 스레드에서 도는 실제 처리. {@code processDelivery} 는 핸들러 예외를 스스로 잡아
+ * RETRY_WAIT 로 적으므로 여기까지 올라오는 것은 그 바깥의 사고(배달 행이 사라졌다든지)뿐이다.
+ * 그것을 삼키지 않고 남긴다 — executor 스레드에서 던지면 아무 데도 안 남고, 그 행은 리스
+ * 만료까지 PROCESSING 으로 보인다.
+ */
+ private void processSafely(String deliveryId) {
+ try {
webhookService.processDelivery(deliveryId);
+ } catch (RuntimeException exception) {
+ log.error("webhook delivery 처리 중 예기치 못한 오류 — 리스 만료 후 회수됩니다. deliveryId={}",
+ deliveryId, exception);
}
}
}
diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml
index e7cff4b5..85f448cd 100644
--- a/src/main/resources/application.yaml
+++ b/src/main/resources/application.yaml
@@ -237,7 +237,12 @@ qeploy:
# ADR-Y2 (#55): StuckApprovalSweeper's poll interval — defense-in-depth recovery for the D1
# write-skew stuck state (structurally prevented by ADR-Y1's lock hierarchy; this is the
# safety net for anything that slips through anyway).
- sweep-interval-ms: ${QEPLOY_AGENT_APPROVAL_SWEEP_INTERVAL_MS:60000}
+ #
+ # #340 5-7: 60초 → 5분. 이 스윕은 자기 자신의 자바독이 "정상이면 항상 0건" 이라고 적은
+ # 회귀망이다. 회귀가 실제로 생겼을 때 5분 안에 잡히면 충분하고(사용자는 이미 승인 버튼을
+ # 누른 뒤 기다리는 중이 아니다 — 그 경로는 ADR-Y1 이 구조적으로 막았다), 그 사이 매분
+ # 도는 것은 정상 상태에서 순수한 낭비다.
+ sweep-interval-ms: ${QEPLOY_AGENT_APPROVAL_SWEEP_INTERVAL_MS:300000}
# AbandonedApprovalSweeper: 아무도 결정하지 않은 승인 대기 태스크를 닫기까지 기다리는 시간.
# 넉넉해야 한다 — 승인 카드를 띄워둔 채 하루 자리를 비우는 것은 정상이고, 그걸 취소하면
# 스윕이 오히려 기능을 망가뜨린다. 0 이하로 두면 스윕이 꺼진다.
@@ -252,6 +257,10 @@ qeploy:
worker:
poll-interval-ms: ${QEPLOY_DEPLOYMENT_WORKER_POLL_INTERVAL_MS:1000}
heartbeat-interval-ms: ${QEPLOY_DEPLOYMENT_WORKER_HEARTBEAT_INTERVAL_MS:30000}
+ # #340 5-2: deploymentExecutor 가 포화라 claim 한 이력을 넘기지 못했을 때, PENDING 으로
+ # 되돌리며 next_run_at 에 얹는 backoff. 포화 상태에서 claim<->release 가 매 폴링 반복되는
+ # 것을 막는다(AgentRunWorker 의 같은 이름 설정과 같은 역할).
+ dispatch-reject-backoff-ms: ${QEPLOY_DEPLOYMENT_WORKER_DISPATCH_REJECT_BACKOFF_MS:5000}
recovery:
# 결과 웹훅을 놓쳐 IN_PROGRESS 에 멈춘 배포를 GitHub 에 직접 물어 회수한다.
poll-interval-ms: ${QEPLOY_DEPLOYMENT_RECOVERY_POLL_INTERVAL_MS:60000}
@@ -292,6 +301,9 @@ qeploy:
webhook:
worker:
poll-interval-ms: ${QEPLOY_WEBHOOK_WORKER_POLL_INTERVAL_MS:1000}
+ # #340 5-3: webhookExecutor 가 포화라 claim 한 배달을 넘기지 못했을 때, PENDING 으로
+ # 되돌리며 next_attempt_at 에 얹는 backoff.
+ dispatch-reject-backoff-ms: ${QEPLOY_WEBHOOK_WORKER_DISPATCH_REJECT_BACKOFF_MS:5000}
# WebhookDeliveryRetentionScheduler (#338): 이 테이블에는 삭제 로직이 없어서 payload
# LONGBLOB 이 이벤트마다 영구히 쌓이고 있었다. 멱등·재시도를 위한 기록이라 최종 상태에
# 도달하고 며칠 지나면 아무도 읽지 않는다. 터미널 상태만 지운다 — 처리 대기·재시도 대기·
@@ -310,6 +322,22 @@ qeploy:
# 사라져 사용자가 인증이 왜 깨졌는지 알 수 없다 — "만료됨"을 얼마간 보여준 뒤 지운다.
expired-grace-days: ${QEPLOY_API_TOKEN_EXPIRED_GRACE_DAYS:30}
+ # #340 5-1 — 1초 폴링 워커(Agent·배포·클라우드 연결·웹훅)의 적응형 백오프.
+ #
+ # 워커의 @Scheduled 간격(각 도메인의 worker.poll-interval-ms)은 그대로 1초이고, 여기 값은
+ # "그 틱에 DB 를 칠지"를 정한다. 일이 있으면 base-delay-ms, 없으면 두 배씩 늘려 max-delay-ms
+ # 까지 물러난다. 유휴 상태 DB 비용의 대부분이 폴링당 트랜잭션 의례였으므로, 폴링 횟수를 줄이면
+ # 의례까지 같은 비율로 줄어든다.
+ #
+ # max-delay-ms 를 늘려도 사용자 대기가 늘지는 않는다 — enqueue 가 커밋되면 그 큐의 백오프가
+ # 즉시 풀린다(WorkerPollGate). 이 값이 실제로 정하는 것은 "다른 인스턴스가 넣은 일을 최대 얼마나
+ # 늦게 보는가"다. 깨우기 신호는 JVM 로컬이라 인스턴스를 넘지 못하므로, 이 폴백 폴링이 다중
+ # 인스턴스 안전성의 마지막 보루다. 0 으로 두면 안 된다.
+ worker:
+ poll:
+ base-delay-ms: ${QEPLOY_WORKER_POLL_BASE_DELAY_MS:1000}
+ max-delay-ms: ${QEPLOY_WORKER_POLL_MAX_DELAY_MS:30000}
+
# Cloudflare DNS - managed subdomain automation
cloudflare:
# 값은 서버 환경변수로만 넣는다 — 이 파일에 적으면 저장소로 새어 나간다.
diff --git a/src/test/java/com/example/dvely/agent/infrastructure/store/TaskStoreTest.java b/src/test/java/com/example/dvely/agent/infrastructure/store/TaskStoreTest.java
index a6691389..7315533f 100644
--- a/src/test/java/com/example/dvely/agent/infrastructure/store/TaskStoreTest.java
+++ b/src/test/java/com/example/dvely/agent/infrastructure/store/TaskStoreTest.java
@@ -70,7 +70,7 @@ void setUp() {
// repository query being exercised against a real DB elsewhere).
when(runRepository.findByTaskIdForUpdate(any(String.class)))
.thenAnswer(invocation -> Optional.ofNullable(runs.get(invocation.getArgument(0))));
- taskStore = new TaskStore(runRepository, eventRepository, new ObjectMapper());
+ taskStore = new TaskStore(runRepository, eventRepository, new ObjectMapper(), event -> { });
}
@Test
@@ -279,7 +279,7 @@ void persistedPlanCanBeReadByNewStoreInstance() {
);
taskStore.savePlan("task-1", plan);
- TaskStore restartedStore = new TaskStore(runRepository, eventRepository, new ObjectMapper());
+ TaskStore restartedStore = new TaskStore(runRepository, eventRepository, new ObjectMapper(), event -> { });
assertThat(restartedStore.getPlan("task-1")).isEqualTo(plan);
}
diff --git a/src/test/java/com/example/dvely/agent/infrastructure/worker/AgentRunWorkerPollingTest.java b/src/test/java/com/example/dvely/agent/infrastructure/worker/AgentRunWorkerPollingTest.java
new file mode 100644
index 00000000..e2b14d55
--- /dev/null
+++ b/src/test/java/com/example/dvely/agent/infrastructure/worker/AgentRunWorkerPollingTest.java
@@ -0,0 +1,174 @@
+package com.example.dvely.agent.infrastructure.worker;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import com.example.dvely.agent.application.orchestrator.AgentPlanExecutor;
+import com.example.dvely.agent.application.service.AgentMessageService;
+import com.example.dvely.agent.infrastructure.store.TaskStore;
+import com.example.dvely.common.worker.WorkQueue;
+import com.example.dvely.common.worker.WorkQueuedEvent;
+import com.example.dvely.common.worker.WorkerPollGate;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.jupiter.api.Test;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+/**
+ * #340 5-1 — 워커가 게이트를 실제로 지키는가. {@code WorkerPollGateTest} 가 백오프 계산 자체를
+ * 못박는다면, 이쪽은 "닫힌 게이트에서는 DB 를 한 번도 치지 않는다"와 "깨우면 바로 친다"를 본다.
+ */
+class AgentRunWorkerPollingTest {
+
+ private static final long BASE_MS = 1_000L;
+ private static final long MAX_MS = 30_000L;
+ private static final long BACKOFF_MS = 5_000L;
+
+ private final AtomicLong nanos = new AtomicLong();
+ private final WorkerPollGate gate = new WorkerPollGate(BASE_MS, MAX_MS, nanos::get);
+ private final TaskStore taskStore = mock(TaskStore.class);
+
+ private void advance(long millis) {
+ nanos.addAndGet(TimeUnit.MILLISECONDS.toNanos(millis));
+ }
+
+ private AgentRunWorker worker() {
+ return new AgentRunWorker(
+ taskStore,
+ mock(AgentPlanExecutor.class),
+ mock(AgentMessageService.class),
+ new AgentExecutionRegistry(),
+ freeExecutor(),
+ gate,
+ BACKOFF_MS
+ );
+ }
+
+ private void stubEmptyPoll() {
+ when(taskStore.recoverAndClaim(anyString(), anyInt()))
+ .thenReturn(new TaskStore.PollBatch(List.of(), List.of()));
+ }
+
+ /**
+ * 이 단위의 목적 그 자체 — 유휴가 이어지면 틱이 와도 쿼리가 나가지 않는다. 유휴 DB
+ * 비용의 대부분이 폴링당 트랜잭션 의례였으므로, 폴링 횟수가 곧 비용이다.
+ */
+ @Test
+ void idleTicksStopTouchingTheDatabaseEntirely() {
+ stubEmptyPoll();
+ AgentRunWorker worker = worker();
+
+ worker.dispatchQueuedRuns(); // 1회차: 실제 폴링, 빈손 → 2초 백오프
+ verify(taskStore, times(1)).recoverAndClaim(anyString(), anyInt());
+
+ // 그 뒤 1초 틱이 아무리 와도 백오프가 끝나기 전에는 쿼리가 없다.
+ advance(1_000);
+ worker.dispatchQueuedRuns();
+ verify(taskStore, times(1)).recoverAndClaim(anyString(), anyInt());
+
+ advance(1_000);
+ worker.dispatchQueuedRuns(); // 2초 경과 — 여기서만 다시 친다
+ verify(taskStore, times(2)).recoverAndClaim(anyString(), anyInt());
+ }
+
+ /**
+ * 첫 claim 지연. 백오프가 상한(30초)까지 늘어난 뒤에도, 태스크가 큐에 들어가면 바로 다음
+ * 틱에 claim 이 나간다 — 사용자가 백오프만큼 기다리는 일은 없다.
+ */
+ @Test
+ void aTaskQueuedWhileFullyBackedOffIsClaimedOnTheVeryNextTick() {
+ stubEmptyPoll();
+ AgentRunWorker worker = worker();
+
+ // 상한까지 물러나게 만든다.
+ for (int i = 0; i < 8; i++) {
+ worker.dispatchQueuedRuns();
+ advance(MAX_MS);
+ }
+ worker.dispatchQueuedRuns();
+ int pollsWhileIdle = org.mockito.Mockito.mockingDetails(taskStore).getInvocations().size();
+ worker.dispatchQueuedRuns(); // 백오프 중 — 쿼리 없음
+ assertThat(org.mockito.Mockito.mockingDetails(taskStore).getInvocations().size())
+ .as("백오프 중인 틱은 DB 를 치지 않는다")
+ .isEqualTo(pollsWhileIdle);
+
+ // 사용자가 메시지를 보냈다 = enqueue 가 커밋됐다.
+ gate.onWorkQueued(new WorkQueuedEvent(WorkQueue.AGENT_RUN));
+
+ worker.dispatchQueuedRuns();
+
+ assertThat(org.mockito.Mockito.mockingDetails(taskStore).getInvocations().size())
+ .as("깨우기 직후 첫 틱에서 claim 이 나가야 한다 — 지연은 최대 폴링 틱 1회(1초)")
+ .isGreaterThan(pollsWhileIdle);
+ }
+
+ /**
+ * 실행기가 포화라 claim 을 생략한 폴링은 "일이 없다"가 아니다. 자리가 나는 것을 알려줄 신호는
+ * 없으므로, 여기서 물러나면 큐에 쌓인 태스크가 백오프 상한만큼 늦게 출발한다.
+ */
+ @Test
+ void aPollSkippedForExecutorSaturationDoesNotBackOff() throws Exception {
+ stubEmptyPoll();
+ ThreadPoolTaskExecutor saturated = new ThreadPoolTaskExecutor();
+ saturated.setCorePoolSize(1);
+ saturated.setMaxPoolSize(1);
+ saturated.setQueueCapacity(0);
+ saturated.initialize();
+ java.util.concurrent.CountDownLatch started = new java.util.concurrent.CountDownLatch(1);
+ java.util.concurrent.CountDownLatch release = new java.util.concurrent.CountDownLatch(1);
+ saturated.execute(() -> {
+ started.countDown();
+ try {
+ release.await();
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ try {
+ started.await(2, TimeUnit.SECONDS);
+ AgentRunWorker worker = new AgentRunWorker(
+ taskStore, mock(AgentPlanExecutor.class), mock(AgentMessageService.class),
+ new AgentExecutionRegistry(), saturated, gate, BACKOFF_MS);
+
+ worker.dispatchQueuedRuns();
+ worker.dispatchQueuedRuns();
+
+ // claim 은 한 번도 안 하지만(claimLimit=0), 회수는 매 틱 돈다 — 물러나지 않았다는 뜻이다.
+ verify(taskStore, times(2)).recoverAndClaim(anyString(), eq(0));
+ } finally {
+ release.countDown();
+ saturated.shutdown();
+ }
+ }
+
+ @Test
+ void aClosedGateSkipsTheStoreCompletely() {
+ stubEmptyPoll();
+ AgentRunWorker worker = worker();
+ worker.dispatchQueuedRuns();
+ org.mockito.Mockito.clearInvocations(taskStore);
+
+ worker.dispatchQueuedRuns(); // 아직 백오프 중
+
+ verifyNoInteractions(taskStore);
+ verify(taskStore, never()).recoverAndClaim(anyString(), anyInt());
+ }
+
+ private ThreadPoolTaskExecutor freeExecutor() {
+ ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+ executor.setCorePoolSize(2);
+ executor.setMaxPoolSize(5);
+ executor.setQueueCapacity(10);
+ executor.initialize();
+ return executor;
+ }
+}
diff --git a/src/test/java/com/example/dvely/agent/infrastructure/worker/AgentRunWorkerTest.java b/src/test/java/com/example/dvely/agent/infrastructure/worker/AgentRunWorkerTest.java
index 1ffaa806..41dc7655 100644
--- a/src/test/java/com/example/dvely/agent/infrastructure/worker/AgentRunWorkerTest.java
+++ b/src/test/java/com/example/dvely/agent/infrastructure/worker/AgentRunWorkerTest.java
@@ -20,9 +20,12 @@
import com.example.dvely.agent.application.service.AgentMessageService;
import com.example.dvely.agent.domain.value.AiProvider;
import com.example.dvely.agent.infrastructure.store.TaskStore;
+import com.example.dvely.common.worker.WorkerPollGate;
import java.time.Instant;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.core.task.TaskRejectedException;
@@ -38,16 +41,17 @@ void recoversLeasesAndDispatchesClaimedTask() {
AgentMessageService messageService = mock(AgentMessageService.class);
AgentPlanExecutor executor = mock(AgentPlanExecutor.class);
AgentExecutionRegistry registry = new AgentExecutionRegistry();
- AgentRunWorker worker = new AgentRunWorker(taskStore, executor, messageService, registry, freeExecutor(), BACKOFF_MS);
+ AgentRunWorker worker = newWorker(taskStore, executor, messageService, registry, freeExecutor(), BACKOFF_MS);
AgentPlan plan = new AgentPlan(List.of(), "reason", AiProvider.OPENAI, 11L);
AgentTask task = agentTask();
- when(taskStore.claimRunnableTasks(anyString(), eq(2))).thenReturn(List.of("task-1"));
+ stubPoll(taskStore, List.of(), List.of("task-1"));
when(taskStore.get("task-1")).thenReturn(task);
when(taskStore.getPlan("task-1")).thenReturn(plan);
worker.dispatchQueuedRuns();
- verify(taskStore).recoverExpiredLeases();
+ // #340 5-1: 회수와 claim 은 이제 한 트랜잭션(recoverAndClaim)이다.
+ verify(taskStore).recoverAndClaim(anyString(), eq(2));
verify(executor).execute(plan, "task-1", 1L);
}
@@ -61,11 +65,10 @@ void leaseExhaustionTellsTheUserInsteadOfLeavingTheScreenSilent() {
TaskStore taskStore = mock(TaskStore.class);
AgentMessageService messageService = mock(AgentMessageService.class);
AgentExecutionRegistry registry = new AgentExecutionRegistry();
- AgentRunWorker worker = new AgentRunWorker(
+ AgentRunWorker worker = newWorker(
taskStore, mock(AgentPlanExecutor.class), messageService, registry, freeExecutor(), BACKOFF_MS);
- when(taskStore.recoverExpiredLeases()).thenReturn(List.of("task-1"));
+ stubPoll(taskStore, List.of("task-1"), List.of());
when(taskStore.get("task-1")).thenReturn(agentTask());
- when(taskStore.claimRunnableTasks(anyString(), eq(2))).thenReturn(List.of());
worker.dispatchQueuedRuns();
@@ -79,10 +82,9 @@ void leaseRecoveryWithoutExhaustionSaysNothing() {
TaskStore taskStore = mock(TaskStore.class);
AgentMessageService messageService = mock(AgentMessageService.class);
AgentExecutionRegistry registry = new AgentExecutionRegistry();
- AgentRunWorker worker = new AgentRunWorker(
+ AgentRunWorker worker = newWorker(
taskStore, mock(AgentPlanExecutor.class), messageService, registry, freeExecutor(), BACKOFF_MS);
- when(taskStore.recoverExpiredLeases()).thenReturn(List.of());
- when(taskStore.claimRunnableTasks(anyString(), eq(2))).thenReturn(List.of());
+ stubPoll(taskStore, List.of(), List.of());
worker.dispatchQueuedRuns();
@@ -97,10 +99,10 @@ void registersTaskInExecutionRegistryBeforeSubmittingToTheExecutor() {
AgentMessageService messageService = mock(AgentMessageService.class);
AgentPlanExecutor executor = mock(AgentPlanExecutor.class);
AgentExecutionRegistry registry = mock(AgentExecutionRegistry.class);
- AgentRunWorker worker = new AgentRunWorker(taskStore, executor, messageService, registry, freeExecutor(), BACKOFF_MS);
+ AgentRunWorker worker = newWorker(taskStore, executor, messageService, registry, freeExecutor(), BACKOFF_MS);
AgentPlan plan = new AgentPlan(List.of(), "reason", AiProvider.OPENAI, 11L);
AgentTask task = agentTask();
- when(taskStore.claimRunnableTasks(anyString(), eq(2))).thenReturn(List.of("task-1"));
+ stubPoll(taskStore, List.of(), List.of("task-1"));
when(taskStore.get("task-1")).thenReturn(task);
when(taskStore.getPlan("task-1")).thenReturn(plan);
@@ -122,11 +124,11 @@ void executorRejectionReleasesClaimUnregistersAndContinuesDispatchingTheRestOfTh
AgentMessageService messageService = mock(AgentMessageService.class);
AgentPlanExecutor executor = mock(AgentPlanExecutor.class);
AgentExecutionRegistry registry = mock(AgentExecutionRegistry.class);
- AgentRunWorker worker = new AgentRunWorker(taskStore, executor, messageService, registry, freeExecutor(), BACKOFF_MS);
+ AgentRunWorker worker = newWorker(taskStore, executor, messageService, registry, freeExecutor(), BACKOFF_MS);
AgentPlan plan = new AgentPlan(List.of(), "reason", AiProvider.OPENAI, 11L);
AgentTask rejectedTask = agentTask("task-1");
AgentTask okTask = agentTask("task-2");
- when(taskStore.claimRunnableTasks(anyString(), eq(2))).thenReturn(List.of("task-1", "task-2"));
+ stubPoll(taskStore, List.of(), List.of("task-1", "task-2"));
when(taskStore.get("task-1")).thenReturn(rejectedTask);
when(taskStore.get("task-2")).thenReturn(okTask);
when(taskStore.getPlan("task-1")).thenReturn(plan);
@@ -152,7 +154,7 @@ void heartbeatSkipsTheRenewalQueryWhenTheRegistryIsEmpty() {
TaskStore taskStore = mock(TaskStore.class);
AgentMessageService messageService = mock(AgentMessageService.class);
AgentExecutionRegistry registry = new AgentExecutionRegistry();
- AgentRunWorker worker = new AgentRunWorker(taskStore, mock(AgentPlanExecutor.class), messageService, registry,
+ AgentRunWorker worker = newWorker(taskStore, mock(AgentPlanExecutor.class), messageService, registry,
freeExecutor(), BACKOFF_MS);
worker.renewLeases();
@@ -167,7 +169,7 @@ void heartbeatRenewsExactlyTheRegisteredTaskIds() {
AgentExecutionRegistry registry = new AgentExecutionRegistry();
registry.register("task-1");
registry.register("task-2");
- AgentRunWorker worker = new AgentRunWorker(taskStore, mock(AgentPlanExecutor.class), messageService, registry,
+ AgentRunWorker worker = newWorker(taskStore, mock(AgentPlanExecutor.class), messageService, registry,
freeExecutor(), BACKOFF_MS);
worker.renewLeases();
@@ -206,19 +208,44 @@ void skipsClaimEntirelyWhenTheExecutorHasNoFreeCapacity() throws InterruptedExce
});
try {
started.await(2, java.util.concurrent.TimeUnit.SECONDS);
- AgentRunWorker worker = new AgentRunWorker(taskStore, mock(AgentPlanExecutor.class), messageService,
+ stubPoll(taskStore, List.of(), List.of());
+ AgentRunWorker worker = newWorker(taskStore, mock(AgentPlanExecutor.class), messageService,
new AgentExecutionRegistry(), saturated, BACKOFF_MS);
worker.dispatchQueuedRuns();
- verify(taskStore).recoverExpiredLeases();
- verify(taskStore, never()).claimRunnableTasks(anyString(), org.mockito.ArgumentMatchers.anyInt());
+ // #340 5-1: 포화여도 회수는 돌아야 한다(좀비 리스가 그만큼 오래 남으므로). 그래서
+ // 폴링을 건너뛰는 대신 claimLimit=0 으로 회수만 시킨다.
+ verify(taskStore).recoverAndClaim(anyString(), eq(0));
} finally {
release.countDown();
saturated.shutdown();
}
}
+ /**
+ * 게이트를 항상 열어 두는 워커. 이 파일의 테스트들은 백오프가 아니라 dispatch 동작을 보므로,
+ * 폴링 호출마다 시계를 넉넉히 흘려 게이트가 판단에 끼어들지 않게 한다. 백오프 자체는
+ * {@code WorkerPollGateTest} 와 {@code AgentRunWorkerBackoffTest} 가 따로 못박는다.
+ */
+ private static AgentRunWorker newWorker(TaskStore taskStore,
+ AgentPlanExecutor planExecutor,
+ AgentMessageService messageService,
+ AgentExecutionRegistry registry,
+ ThreadPoolTaskExecutor agentExecutor,
+ long backoffMs) {
+ AtomicLong nanos = new AtomicLong();
+ WorkerPollGate openGate = new WorkerPollGate(
+ 1000L, 30_000L, () -> nanos.addAndGet(TimeUnit.MINUTES.toNanos(1)));
+ return new AgentRunWorker(
+ taskStore, planExecutor, messageService, registry, agentExecutor, openGate, backoffMs);
+ }
+
+ private static void stubPoll(TaskStore taskStore, List leaseExhausted, List claimed) {
+ when(taskStore.recoverAndClaim(anyString(), org.mockito.ArgumentMatchers.anyInt()))
+ .thenReturn(new TaskStore.PollBatch(leaseExhausted, claimed));
+ }
+
private String workerIdOf(AgentRunWorker worker) {
return java.lang.management.ManagementFactory.getRuntimeMXBean().getName();
}
diff --git a/src/test/java/com/example/dvely/agent/infrastructure/worker/DockerGarbageSweeperTest.java b/src/test/java/com/example/dvely/agent/infrastructure/worker/DockerGarbageSweeperTest.java
index d54b069e..391da0d8 100644
--- a/src/test/java/com/example/dvely/agent/infrastructure/worker/DockerGarbageSweeperTest.java
+++ b/src/test/java/com/example/dvely/agent/infrastructure/worker/DockerGarbageSweeperTest.java
@@ -17,7 +17,11 @@ class DockerGarbageSweeperTest {
private final DockerContainerService dockerService = mock(DockerContainerService.class);
private DockerGarbageSweeper sweeper(boolean enabled, long keepHours) {
- DockerGarbageSweeper sweeper = new DockerGarbageSweeper(dockerService);
+ return sweeper(enabled, keepHours, Runnable::run);
+ }
+
+ private DockerGarbageSweeper sweeper(boolean enabled, long keepHours, java.util.concurrent.Executor executor) {
+ DockerGarbageSweeper sweeper = new DockerGarbageSweeper(dockerService, executor);
ReflectionTestUtils.setField(sweeper, "enabled", enabled);
ReflectionTestUtils.setField(sweeper, "buildCacheKeepHours", keepHours);
return sweeper;
@@ -51,4 +55,21 @@ void survivesWhenDockerIsUnreachable() {
Assertions.assertThatCode(() -> sweeper(true, 48).sweep()).doesNotThrowAnyException();
}
+
+ /**
+ * #340 5-10 — prune 은 스케줄러 스레드에서 돌지 않는다. prune 3회가 도커 데몬을 잠시
+ * 붙잡는 동안 스케줄러를 점유하면 같은 풀을 쓰는 다른 잡이 그만큼 밀린다.
+ */
+ @Test
+ void pruningRunsOnTheMaintenanceExecutorNotTheSchedulerThread() {
+ java.util.List submitted = new java.util.ArrayList<>();
+
+ sweeper(true, 48, submitted::add).sweep();
+
+ Assertions.assertThat(submitted).hasSize(1);
+ verify(dockerService, never()).pruneGarbage(any());
+
+ submitted.get(0).run();
+ verify(dockerService).pruneGarbage(Duration.ofHours(48));
+ }
}
diff --git a/src/test/java/com/example/dvely/chat/application/command/ChatCommandServiceTest.java b/src/test/java/com/example/dvely/chat/application/command/ChatCommandServiceTest.java
index ff6ec0ed..5d41807c 100644
--- a/src/test/java/com/example/dvely/chat/application/command/ChatCommandServiceTest.java
+++ b/src/test/java/com/example/dvely/chat/application/command/ChatCommandServiceTest.java
@@ -198,32 +198,20 @@ void permanentlyDeleteConversationRejectsActiveConversation() {
verify(conversationRepository, never()).deleteById(any());
}
+ /**
+ * #340 5-9: 만료된 휴지통 대화를 벌크 DELETE 한 문장으로 지운다. 예전에는 엔티티를 전부
+ * 로드한 뒤 {@code deleteById} 를 N 번 불렀다 — 지우려고 읽고, 지우려고 또 왕복했다.
+ * 삭제 조건이 곧 SELECT 조건이었으므로 읽을 이유가 없다.
+ */
@Test
- void purgeExpiredConversationsDeletesRepositoryMatches() {
- Conversation first = new Conversation(
- 11L,
- 2L,
- 7L,
- true,
- LocalDateTime.now().minusDays(8),
- LocalDateTime.now().minusDays(10),
- LocalDateTime.now().minusDays(8)
- );
- Conversation second = new Conversation(
- 12L,
- 2L,
- 7L,
- true,
- LocalDateTime.now().minusDays(9),
- LocalDateTime.now().minusDays(10),
- LocalDateTime.now().minusDays(9)
- );
- when(conversationRepository.findAllByDeletedTrueAndDeletedAtLessThanEqual(any()))
- .thenReturn(List.of(first, second));
+ void purgeExpiredConversationsDeletesInOneBulkStatement() {
+ when(conversationRepository.deleteExpiredTrash(any())).thenReturn(2);
assertThat(chatCommandService.purgeExpiredConversations()).isEqualTo(2);
- verify(conversationRepository).deleteById(11L);
- verify(conversationRepository).deleteById(12L);
+
+ verify(conversationRepository).deleteExpiredTrash(any());
+ verify(conversationRepository, never()).findAllByDeletedTrueAndDeletedAtLessThanEqual(any());
+ verify(conversationRepository, never()).deleteById(any());
}
private Project project(Long projectId, Long ownerUserId, String sourceRepository, boolean deleted) {
diff --git a/src/test/java/com/example/dvely/cloudconnection/infrastructure/worker/CloudConnectionVerificationWorkerTest.java b/src/test/java/com/example/dvely/cloudconnection/infrastructure/worker/CloudConnectionVerificationWorkerTest.java
new file mode 100644
index 00000000..058e97c5
--- /dev/null
+++ b/src/test/java/com/example/dvely/cloudconnection/infrastructure/worker/CloudConnectionVerificationWorkerTest.java
@@ -0,0 +1,65 @@
+package com.example.dvely.cloudconnection.infrastructure.worker;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+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.example.dvely.cloudconnection.application.service.CloudConnectionVerificationService;
+import com.example.dvely.cloudconnection.domain.repository.CloudConnectionVerificationJobRepository;
+import com.example.dvely.common.worker.WorkerPollGate;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.jupiter.api.Test;
+import org.springframework.core.task.TaskRejectedException;
+
+class CloudConnectionVerificationWorkerTest {
+
+ /** 백오프가 이 파일의 dispatch 검증에 끼어들지 않도록 게이트를 항상 열어 둔다. */
+ private static WorkerPollGate openGate() {
+ AtomicLong nanos = new AtomicLong();
+ return new WorkerPollGate(1000L, 30_000L, () -> nanos.addAndGet(TimeUnit.MINUTES.toNanos(1)));
+ }
+
+ @Test
+ void dispatchPendingJobs_recoversClaimsAndDelegatesJobs() {
+ CloudConnectionVerificationJobRepository repository =
+ mock(CloudConnectionVerificationJobRepository.class);
+ CloudConnectionVerificationService service = mock(CloudConnectionVerificationService.class);
+ CloudConnectionVerificationWorker worker = new CloudConnectionVerificationWorker(repository, service, openGate());
+ when(repository.recoverAndClaimPending(anyString(), eq(2))).thenReturn(List.of("job-1", "job-2"));
+
+ worker.dispatchPendingJobs();
+
+ // #340 5-1: 회수와 claim 은 이제 한 트랜잭션(recoverAndClaimPending)이다.
+ verify(repository).recoverAndClaimPending(anyString(), eq(2));
+ verify(service).executeQueued("job-1");
+ verify(service).executeQueued("job-2");
+ }
+
+ /**
+ * #340 5-2 재현: {@code cloudConnectionExecutor} 포화로 첫 job 의 위임이 거부되면, 예전에는 그
+ * 예외가 루프를 끊어 같은 배치의 job-2 가 RUNNING 인 채 리스 만료(2분)까지 방치됐다. 그 동안
+ * 사용자에게는 "권한을 확인하고 있습니다"만 떠 있고 실제로 확인하는 것은 아무것도 없었다.
+ */
+ @Test
+ void executorRejectionDoesNotStrandTheRestOfTheClaimedBatch() {
+ CloudConnectionVerificationJobRepository repository =
+ mock(CloudConnectionVerificationJobRepository.class);
+ CloudConnectionVerificationService service = mock(CloudConnectionVerificationService.class);
+ CloudConnectionVerificationWorker worker = new CloudConnectionVerificationWorker(repository, service, openGate());
+ when(repository.recoverAndClaimPending(anyString(), eq(2))).thenReturn(List.of("job-1", "job-2"));
+ doThrow(new TaskRejectedException("cloudConnectionExecutor 포화"))
+ .when(service).executeQueued("job-1");
+
+ worker.dispatchPendingJobs();
+
+ verify(service).executeQueued("job-2");
+ verify(repository).releaseClaim(eq("job-1"), anyString());
+ verify(repository, never()).releaseClaim(eq("job-2"), anyString());
+ }
+}
diff --git a/src/test/java/com/example/dvely/common/worker/WorkerPollGateTest.java b/src/test/java/com/example/dvely/common/worker/WorkerPollGateTest.java
new file mode 100644
index 00000000..f6424d87
--- /dev/null
+++ b/src/test/java/com/example/dvely/common/worker/WorkerPollGateTest.java
@@ -0,0 +1,125 @@
+package com.example.dvely.common.worker;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.jupiter.api.Test;
+
+/**
+ * #340 5-1 — 적응형 백오프의 계약.
+ *
+ * 시계를 주입해 벽시계에 의존하지 않는다. 백오프는 "몇 초 뒤"가 본질이라 실제로 재우면
+ * 테스트가 느려지고 흔들린다.
+ */
+class WorkerPollGateTest {
+
+ private static final long BASE_MS = 1_000L;
+ private static final long MAX_MS = 30_000L;
+
+ private final AtomicLong nanos = new AtomicLong();
+ private final WorkerPollGate gate = new WorkerPollGate(BASE_MS, MAX_MS, nanos::get);
+
+ private void advance(long millis) {
+ nanos.addAndGet(TimeUnit.MILLISECONDS.toNanos(millis));
+ }
+
+ @Test
+ void firstTickAfterStartupAlwaysPolls() {
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN)).isTrue();
+ }
+
+ @Test
+ void idlePollsBackOffByDoublingUpToTheCap() {
+ // 유휴가 이어지면 간격이 1 → 2 → 4 → 8 → 16 → 30(상한) 초로 늘어난다.
+ long[] expectedDelaysMs = {2_000L, 4_000L, 8_000L, 16_000L, 30_000L, 30_000L};
+ for (long expected : expectedDelaysMs) {
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN)).isTrue();
+ gate.recordIdle(WorkQueue.AGENT_RUN);
+
+ // 기대 간격 직전까지는 닫혀 있고, 지나면 열린다.
+ advance(expected - 1);
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN))
+ .as("%dms 백오프가 끝나기 전에는 DB 를 치지 않는다", expected)
+ .isFalse();
+ advance(1);
+ }
+ }
+
+ @Test
+ void findingWorkResetsTheIntervalToTheMinimum() {
+ gate.shouldPoll(WorkQueue.AGENT_RUN);
+ gate.recordIdle(WorkQueue.AGENT_RUN);
+ gate.shouldPoll(WorkQueue.AGENT_RUN);
+ advance(2_000);
+ gate.shouldPoll(WorkQueue.AGENT_RUN);
+ gate.recordIdle(WorkQueue.AGENT_RUN); // 여기서 간격은 4초
+
+ advance(4_000);
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN)).isTrue();
+ gate.recordBusy(WorkQueue.AGENT_RUN);
+
+ // 일을 찾았으니 즉시 다시 열려 있어야 한다 — 큐에 더 남아 있을 수 있다.
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN)).isTrue();
+ gate.recordIdle(WorkQueue.AGENT_RUN);
+ advance(1_999);
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN))
+ .as("간격이 최소(1초)로 초기화됐으므로 다음 백오프는 4초가 아니라 2초다")
+ .isFalse();
+ advance(1);
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN)).isTrue();
+ }
+
+ /**
+ * 이 단위의 핵심 계약 — 백오프가 상한까지 늘어나 있어도, 일이 들어오면 즉시 열린다.
+ * 이것이 없으면 사용자가 메시지를 보낸 뒤 최대 30초를 기다리게 된다.
+ */
+ @Test
+ void wakingUpOpensTheGateImmediatelyNoMatterHowFarItHadBackedOff() {
+ for (int i = 0; i < 10; i++) {
+ gate.shouldPoll(WorkQueue.AGENT_RUN);
+ gate.recordIdle(WorkQueue.AGENT_RUN);
+ advance(MAX_MS);
+ }
+ gate.shouldPoll(WorkQueue.AGENT_RUN);
+ gate.recordIdle(WorkQueue.AGENT_RUN); // 상한(30초) 백오프에 들어간 상태
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN)).isFalse();
+
+ gate.onWorkQueued(new WorkQueuedEvent(WorkQueue.AGENT_RUN));
+
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN))
+ .as("enqueue 커밋 뒤에는 백오프 값과 무관하게 다음 틱에 폴링해야 한다")
+ .isTrue();
+ }
+
+ @Test
+ void wakingOneQueueDoesNotDisturbAnother() {
+ gate.shouldPoll(WorkQueue.AGENT_RUN);
+ gate.recordIdle(WorkQueue.AGENT_RUN);
+ gate.shouldPoll(WorkQueue.WEBHOOK_DELIVERY);
+ gate.recordIdle(WorkQueue.WEBHOOK_DELIVERY);
+
+ gate.onWorkQueued(new WorkQueuedEvent(WorkQueue.AGENT_RUN));
+
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN)).isTrue();
+ assertThat(gate.shouldPoll(WorkQueue.WEBHOOK_DELIVERY))
+ .as("배포 큐에 일이 들어왔다고 웹훅 워커까지 깨울 이유는 없다")
+ .isFalse();
+ }
+
+ /**
+ * 폴링이 도는 도중에 깨우기가 오는 경우. 그 폴링은 방금 커밋된 행을 못 봤을 수 있으므로,
+ * 빈손으로 끝났다고 백오프를 늘리면 이미 들어온 일이 최대 상한만큼 늦어진다.
+ */
+ @Test
+ void aWakeupThatArrivesMidPollIsNotSwallowedByThatPollsBackoff() {
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN)).isTrue(); // 폴링 시작
+
+ gate.onWorkQueued(new WorkQueuedEvent(WorkQueue.AGENT_RUN)); // 폴링 도중 enqueue 커밋
+ gate.recordIdle(WorkQueue.AGENT_RUN); // 이 폴링은 빈손이었다
+
+ assertThat(gate.shouldPoll(WorkQueue.AGENT_RUN))
+ .as("폴링 중에 온 깨우기는 그 폴링의 백오프에 묻히면 안 된다")
+ .isTrue();
+ }
+}
diff --git a/src/test/java/com/example/dvely/deployment/application/command/DeploymentCommandServiceTest.java b/src/test/java/com/example/dvely/deployment/application/command/DeploymentCommandServiceTest.java
index 409e643e..fec54112 100644
--- a/src/test/java/com/example/dvely/deployment/application/command/DeploymentCommandServiceTest.java
+++ b/src/test/java/com/example/dvely/deployment/application/command/DeploymentCommandServiceTest.java
@@ -79,6 +79,7 @@ void setUp() {
githubActionsPort,
githubRepoPort,
deploymentHistoryRepository,
+ new com.example.dvely.deployment.infrastructure.worker.DeploymentExecutionRegistry(),
policyRepository,
resultApprovalService,
auditRecorder,
diff --git a/src/test/java/com/example/dvely/deployment/infrastructure/worker/DeploymentRunWorkerTest.java b/src/test/java/com/example/dvely/deployment/infrastructure/worker/DeploymentRunWorkerTest.java
index 7e8b5f1f..c2764b85 100644
--- a/src/test/java/com/example/dvely/deployment/infrastructure/worker/DeploymentRunWorkerTest.java
+++ b/src/test/java/com/example/dvely/deployment/infrastructure/worker/DeploymentRunWorkerTest.java
@@ -1,28 +1,145 @@
package com.example.dvely.deployment.infrastructure.worker;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
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.example.dvely.deployment.application.command.DeploymentCommandService;
+import com.example.dvely.common.worker.WorkerPollGate;
import com.example.dvely.deployment.domain.repository.DeploymentHistoryRepository;
+import java.util.Collection;
import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import static org.assertj.core.api.Assertions.assertThat;
+
import org.junit.jupiter.api.Test;
+import org.springframework.core.task.TaskRejectedException;
class DeploymentRunWorkerTest {
+ private static final long BACKOFF_MS = 5000L;
+
+ /** 백오프가 이 파일의 dispatch 검증에 끼어들지 않도록 게이트를 항상 열어 둔다. */
+ private static WorkerPollGate openGate() {
+ AtomicLong nanos = new AtomicLong();
+ return new WorkerPollGate(1000L, 30_000L, () -> nanos.addAndGet(TimeUnit.MINUTES.toNanos(1)));
+ }
+
@Test
void dispatchPendingDeployments_recoversClaimsAndDelegatesJobs() {
DeploymentHistoryRepository repository = mock(DeploymentHistoryRepository.class);
DeploymentCommandService commandService = mock(DeploymentCommandService.class);
- DeploymentRunWorker worker = new DeploymentRunWorker(repository, commandService);
- when(repository.claimPending(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq(2)))
- .thenReturn(List.of(51L, 52L));
+ DeploymentExecutionRegistry registry = new DeploymentExecutionRegistry();
+ DeploymentRunWorker worker = new DeploymentRunWorker(repository, commandService, openGate(), registry, BACKOFF_MS);
+ when(repository.recoverAndClaimPending(anyString(), eq(2))).thenReturn(List.of(51L, 52L));
worker.dispatchPendingDeployments();
- verify(repository).recoverExpiredLeases();
+ // #340 5-1: 회수와 claim 은 이제 한 트랜잭션(recoverAndClaimPending)이다.
+ verify(repository).recoverAndClaimPending(anyString(), eq(2));
verify(commandService).executeQueued(51L);
verify(commandService).executeQueued(52L);
}
+
+ /**
+ * #340 5-2 재현: {@code deploymentExecutor} 포화로 첫 이력의 위임이 거부되면, 예전에는 그
+ * 예외가 dispatch 루프를 통째로 끊어 같은 배치에서 이미 claim 된 52 번이 영영 실행되지
+ * 않았다. 그 행은 IN_PROGRESS 인 채 리스 만료(2분)까지 남고 사용자 화면에는 "배포 중"이 계속
+ * 떠 있었다. 두 가지를 함께 못박는다 — 뒤 이력은 그대로 위임되고, 거부된 이력은 claim 을
+ * 붙든 채 방치되지 않는다.
+ */
+ @Test
+ void executorRejectionDoesNotStrandTheRestOfTheClaimedBatch() {
+ DeploymentHistoryRepository repository = mock(DeploymentHistoryRepository.class);
+ DeploymentCommandService commandService = mock(DeploymentCommandService.class);
+ DeploymentExecutionRegistry registry = new DeploymentExecutionRegistry();
+ DeploymentRunWorker worker = new DeploymentRunWorker(repository, commandService, openGate(), registry, BACKOFF_MS);
+ when(repository.recoverAndClaimPending(anyString(), eq(2))).thenReturn(List.of(51L, 52L));
+ doThrow(new TaskRejectedException("deploymentExecutor 포화"))
+ .when(commandService).executeQueued(51L);
+
+ worker.dispatchPendingDeployments();
+
+ verify(commandService).executeQueued(52L);
+ verify(repository).releaseClaim(eq(51L), anyString(), eq(BACKOFF_MS));
+ verify(repository, never()).releaseClaim(eq(52L), anyString(), anyLong());
+ }
+
+ /** 제출 전 어떤 예외든 결론은 같다 — 이 이력은 돌지 않으므로 claim 을 붙들고 있으면 안 된다. */
+ @Test
+ void anyPreSubmissionFailureReleasesTheClaimToo() {
+ DeploymentHistoryRepository repository = mock(DeploymentHistoryRepository.class);
+ DeploymentCommandService commandService = mock(DeploymentCommandService.class);
+ DeploymentExecutionRegistry registry = new DeploymentExecutionRegistry();
+ DeploymentRunWorker worker = new DeploymentRunWorker(repository, commandService, openGate(), registry, BACKOFF_MS);
+ when(repository.recoverAndClaimPending(anyString(), eq(2))).thenReturn(List.of(51L));
+ doThrow(new IllegalStateException("제출 전 예외")).when(commandService).executeQueued(51L);
+
+ worker.dispatchPendingDeployments();
+
+ verify(repository).releaseClaim(eq(51L), anyString(), eq(BACKOFF_MS));
+ }
+
+ // ── #340 5-8: 하트비트를 실제 실행 중인 이력으로 좁힌다 ──────────────────────────────────
+
+ @Test
+ void heartbeatSkipsTheRenewalQueryWhenNothingIsExecuting() {
+ DeploymentHistoryRepository repository = mock(DeploymentHistoryRepository.class);
+ DeploymentCommandService commandService = mock(DeploymentCommandService.class);
+ DeploymentExecutionRegistry registry = new DeploymentExecutionRegistry();
+ DeploymentRunWorker worker =
+ new DeploymentRunWorker(repository, commandService, openGate(), registry, BACKOFF_MS);
+
+ worker.renewLeases();
+
+ // 실행 중인 배포가 없는데 0행짜리 UPDATE 를 30초마다 내보낼 이유가 없다.
+ verify(repository, never()).renewLeases(anyString(), org.mockito.ArgumentMatchers.anyCollection());
+ }
+
+ @Test
+ void heartbeatRenewsExactlyTheRegisteredHistoryIds() {
+ DeploymentHistoryRepository repository = mock(DeploymentHistoryRepository.class);
+ DeploymentCommandService commandService = mock(DeploymentCommandService.class);
+ DeploymentExecutionRegistry registry = new DeploymentExecutionRegistry();
+ registry.register(51L);
+ registry.register(52L);
+ DeploymentRunWorker worker =
+ new DeploymentRunWorker(repository, commandService, openGate(), registry, BACKOFF_MS);
+
+ worker.renewLeases();
+
+ @SuppressWarnings("unchecked")
+ org.mockito.ArgumentCaptor> captor =
+ org.mockito.ArgumentCaptor.forClass(Collection.class);
+ verify(repository).renewLeases(anyString(), captor.capture());
+ assertThat(captor.getValue()).containsExactlyInAnyOrder(51L, 52L);
+ }
+
+ /**
+ * 실행기가 거부한 이력은 하트비트 대상에서 빠져야 한다. 안 빠지면 실행 주체가 없는 행의
+ * 리스를 30초마다 되살리게 되고, recoverExpiredLeases 가 그것을 영영 회수하지 못한다.
+ */
+ @Test
+ void aRejectedDispatchLeavesNothingBehindForTheHeartbeatToRenew() {
+ DeploymentHistoryRepository repository = mock(DeploymentHistoryRepository.class);
+ DeploymentCommandService commandService = mock(DeploymentCommandService.class);
+ DeploymentExecutionRegistry registry = new DeploymentExecutionRegistry();
+ DeploymentRunWorker worker =
+ new DeploymentRunWorker(repository, commandService, openGate(), registry, BACKOFF_MS);
+ when(repository.recoverAndClaimPending(anyString(), eq(2))).thenReturn(List.of(51L));
+ doThrow(new TaskRejectedException("deploymentExecutor 포화"))
+ .when(commandService).executeQueued(51L);
+
+ worker.dispatchPendingDeployments();
+
+ assertThat(registry.snapshot()).isEmpty();
+ worker.renewLeases();
+ verify(repository, never()).renewLeases(anyString(), org.mockito.ArgumentMatchers.anyCollection());
+ }
}
diff --git a/src/test/java/com/example/dvely/deployment/infrastructure/worker/StuckDeploymentRecoveryWorkerTest.java b/src/test/java/com/example/dvely/deployment/infrastructure/worker/StuckDeploymentRecoveryWorkerTest.java
index 8d66680d..5713e653 100644
--- a/src/test/java/com/example/dvely/deployment/infrastructure/worker/StuckDeploymentRecoveryWorkerTest.java
+++ b/src/test/java/com/example/dvely/deployment/infrastructure/worker/StuckDeploymentRecoveryWorkerTest.java
@@ -5,6 +5,7 @@
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -180,6 +181,45 @@ void oneHistoryFailingDoesNotAbortTheRestOfTheBatch() {
verify(outcomeService).applySuccess(healthy, project);
}
+ /**
+ * #340 5-4 — 판정을 못 얻은 이력을 매 주기 다시 묻지 않는다.
+ *
+ * 예전에는 멈춘 배포 한 건이 포기 시각(기본 120분)에 닿을 때까지 매분 GitHub 을
+ * 쳤다 — 한 건당 최대 120회다. GitHub Actions 실행 상태는 초 단위로 바뀌지 않으므로 결론이
+ * 없는 조회가 이어지면 물러난다(1 → 2 → 5 → 10분).
+ */
+ @Test
+ void anInconclusiveLookupIsNotRepeatedOnTheVeryNextSweep() {
+ // 아직 도는 중 = 판정 없음. updatedAt 이 최근이라 포기 시각(120분)에도 닿지 않았다.
+ DeploymentHistory history = stuckHistory(901L, LocalDateTime.now().minusMinutes(20));
+ givenStuck(history);
+ when(projectRepository.findById(11L)).thenReturn(Optional.of(project()));
+ givenActiveToken();
+ when(githubActionsPort.getWorkflowRunStatus("user-token", "octo/repo", 901L))
+ .thenReturn(new GithubActionsPort.WorkflowRunStatus(901L, "in_progress", null));
+
+ worker.recoverStuckDeployments();
+ worker.recoverStuckDeployments();
+ worker.recoverStuckDeployments();
+
+ // 세 번 돌았지만 GitHub 은 한 번만 물었다 — 나머지는 백오프가 걸러냈다.
+ verify(githubActionsPort, times(1))
+ .getWorkflowRunStatus("user-token", "octo/repo", 901L);
+ verify(outcomeService, never()).applyFailure(any(), any(), any(), anyString());
+ }
+
+ /** 프로젝트가 사라진 이력도 매분 다시 확인할 이유가 없다 — 다음 주기에도 없다. */
+ @Test
+ void aHistoryWhoseProjectVanishedIsNotRecheckedEverySweep() {
+ givenStuck(stuckHistory(902L, LocalDateTime.now().minusMinutes(20)));
+ when(projectRepository.findById(11L)).thenReturn(Optional.empty());
+
+ worker.recoverStuckDeployments();
+ worker.recoverStuckDeployments();
+
+ verify(projectRepository, times(1)).findById(11L);
+ }
+
private void givenStuck(DeploymentHistory history) {
when(historyRepository.findDispatchedAwaitingOutcome(any(), eq(20)))
.thenReturn(List.of(history));
diff --git a/src/test/java/com/example/dvely/domainbinding/infrastructure/worker/DomainVerificationWorkerTest.java b/src/test/java/com/example/dvely/domainbinding/infrastructure/worker/DomainVerificationWorkerTest.java
index 13b71c8f..5cdac934 100644
--- a/src/test/java/com/example/dvely/domainbinding/infrastructure/worker/DomainVerificationWorkerTest.java
+++ b/src/test/java/com/example/dvely/domainbinding/infrastructure/worker/DomainVerificationWorkerTest.java
@@ -35,7 +35,10 @@ void setUp() {
worker = new DomainVerificationWorker(
domainBindingRepository,
commandService,
- new DomainVerificationProperties(60000L, 20, 30, 1440)
+ new DomainVerificationProperties(60000L, 20, 30, 1440),
+ // 검증을 호출 스레드에서 그대로 돌린다 — 이 파일이 보는 것은 "무엇을 검증했는가"이지
+ // 어느 스레드에서 돌았는가가 아니다(#340 5-5 로 실서비스는 전용 executor 를 쓴다).
+ Runnable::run
);
}
@@ -54,6 +57,23 @@ void verifiesEveryVerifyingDomain() {
verify(commandService).checkVerificationAsSystem(2L);
}
+ /**
+ * #340 5-5 — 갓 만든 도메인이라도 매 주기(60초) 다시 검증하지는 않는다. 커스텀 도메인은
+ * TTL 이 1440분이라 예전에는 한 건당 최대 1,440회, 그것도 Cloudflare + GitHub + HTTPS 프로브
+ * 세 묶음이 나갔다. 기다리는 대상이 DNS 전파와 호스팅 반영이라 그렇게 촘촘히 볼 이유가 없다.
+ */
+ @Test
+ void aDomainCheckedJustNowIsNotRecheckedOnTheVeryNextSweep() {
+ givenVerifyingDomains(managedSubdomain(1L, "a.qeploy.com", LocalDateTime.now()));
+ when(commandService.checkVerificationAsSystem(anyLong())).thenReturn(result(DomainStatus.VERIFYING));
+
+ worker.verifyPendingDomains();
+ worker.verifyPendingDomains();
+ worker.verifyPendingDomains();
+
+ verify(commandService, org.mockito.Mockito.times(1)).checkVerificationAsSystem(1L);
+ }
+
@Test
void oneDomainFailingDoesNotAbortTheRestOfTheBatch() {
// Cloudflare·GitHub 호출이 섞여 있어 어느 하나는 언제든 실패한다. 다음 주기에 다시
diff --git a/src/test/java/com/example/dvely/webhook/WebhookMergedPollClaimIntegrationTest.java b/src/test/java/com/example/dvely/webhook/WebhookMergedPollClaimIntegrationTest.java
new file mode 100644
index 00000000..38c43a36
--- /dev/null
+++ b/src/test/java/com/example/dvely/webhook/WebhookMergedPollClaimIntegrationTest.java
@@ -0,0 +1,227 @@
+package com.example.dvely.webhook;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.example.dvely.common.worker.WorkQueue;
+import com.example.dvely.common.worker.WorkQueuedEvent;
+import com.example.dvely.common.worker.WorkerPollGate;
+import com.example.dvely.webhook.domain.model.WebhookDelivery;
+import com.example.dvely.webhook.domain.repository.WebhookDeliveryRepository;
+import com.example.dvely.webhook.domain.value.WebhookDeliveryStatus;
+import java.nio.charset.StandardCharsets;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.support.TransactionTemplate;
+
+/**
+ * #340 5-1 — 폴링 구조를 바꾸고도 깨지지 않아야 하는 두 가지를 실제 MySQL 로 증명한다.
+ *
+ * 1. claim 의 원자성. 회수와 claim 을 한 트랜잭션으로 합쳤어도, 두 워커 인스턴스가
+ * 같은 행을 동시에 노리면 정확히 하나만 집는다. 이 성질은 "후보를 SELECT 한 뒤 행 단위 조건부
+ * UPDATE 로 집고, 영향받은 행 수로 성공을 판정한다"는 구조에서 나온다 — 합치기는 그 구조를
+ * 건드리지 않으므로 성질도 그대로다. 말로 그렇다는 것과 실제로 그런 것은 다르므로 두 스레드로
+ * 재현한다.
+ *
+ * 2. 깨우기의 트랜잭션 경계. 신호는 커밋 뒤에 도착해야 한다. 커밋 전에 깨우면
+ * 워커가 아직 보이지 않는 행을 찾다가 빈손으로 돌아가 도로 백오프에 들어가고, 그 일은 백오프
+ * 상한만큼 늦어진다.
+ *
+ * 이 클래스가 웹훅 큐를 쓰는 이유는 {@code build.gradle} 이 테스트 JVM 전체에서 웹훅 워커의
+ * 폴링을 1시간으로 꺼두기 때문이다 — 살아 있는 워커가 끼어들어 게이트 상태와 행을 흔들지 않는
+ * 유일한 큐다.
+ */
+@SpringBootTest
+class WebhookMergedPollClaimIntegrationTest {
+
+ /** 공유 스키마에 남은 오래된 행에 밀려 갓 심은 행이 배치에서 빠지지 않도록 넉넉히 잡는다. */
+ private static final int POLL_LIMIT = 5_000;
+
+ @Autowired
+ private WebhookDeliveryRepository webhookDeliveryRepository;
+ @Autowired
+ private WorkerPollGate pollGate;
+ @Autowired
+ private ApplicationEventPublisher eventPublisher;
+ @Autowired
+ private TransactionTemplate transactionTemplate;
+ @Autowired
+ private JdbcTemplate jdbcTemplate;
+
+ // ── 1. claim 원자성 ────────────────────────────────────────────────────────────────────────
+
+ @Test
+ void mergedPollRecoversAndClaimsInOneCall() {
+ String deliveryId = uniqueId("merged-basic");
+ seedDelivery(deliveryId, WebhookDeliveryStatus.PENDING, LocalDateTime.now().minusSeconds(1), null, null);
+
+ List claimed = webhookDeliveryRepository.recoverAndClaimPending("worker-a", POLL_LIMIT);
+
+ assertThat(claimed).contains(deliveryId);
+ Map row = fetchRow(deliveryId);
+ assertThat(row.get("status")).isEqualTo(WebhookDeliveryStatus.PROCESSING.name());
+ assertThat(row.get("lease_owner")).isEqualTo("worker-a");
+ }
+
+ @Test
+ void concurrentMergedPollsFromTwoWorkerInstancesNeverBothWinTheSameDelivery() throws Exception {
+ String deliveryId = uniqueId("merged-race");
+ seedDelivery(deliveryId, WebhookDeliveryStatus.PENDING, LocalDateTime.now().minusSeconds(1), null, null);
+
+ CountDownLatch startBarrier = new CountDownLatch(2);
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ List>> futures;
+ try {
+ futures = pool.invokeAll(List.of(
+ racePoll(startBarrier, "worker-merged-a"),
+ racePoll(startBarrier, "worker-merged-b")
+ ));
+ } finally {
+ pool.shutdown();
+ }
+
+ long winners = 0;
+ for (Future> future : futures) {
+ if (future.get(20, TimeUnit.SECONDS).contains(deliveryId)) {
+ winners++;
+ }
+ }
+ assertThat(winners)
+ .as("회수와 claim 을 한 트랜잭션으로 합쳐도 같은 행을 두 인스턴스가 집으면 안 된다")
+ .isEqualTo(1);
+ Map row = fetchRow(deliveryId);
+ assertThat(row.get("status")).isEqualTo(WebhookDeliveryStatus.PROCESSING.name());
+ assertThat(row.get("lease_owner")).isIn("worker-merged-a", "worker-merged-b");
+ }
+
+ @Test
+ void alreadyClaimedRowIsNotReselectedByAnotherInstancesMergedPoll() {
+ String deliveryId = uniqueId("merged-no-reselect");
+ seedDelivery(deliveryId, WebhookDeliveryStatus.PENDING, LocalDateTime.now().minusSeconds(1), null, null);
+
+ assertThat(webhookDeliveryRepository.recoverAndClaimPending("worker-a", POLL_LIMIT)).contains(deliveryId);
+ assertThat(webhookDeliveryRepository.recoverAndClaimPending("worker-b", POLL_LIMIT))
+ .doesNotContain(deliveryId);
+ assertThat(fetchRow(deliveryId).get("lease_owner")).isEqualTo("worker-a");
+ }
+
+ /**
+ * 회수가 합친 뒤에도 실제로 돌아가는지 — 합치면서 조용히 빠뜨리면 좀비 리스가 영원히 남는다.
+ *
+ * 합쳤기 때문에 두 결과가 모두 정상이다: 회수만 되어 RETRY_WAIT 로 남거나, 회수된
+ * 행을 같은 호출의 claim 이 곧바로 다시 집거나. {@code recoverExpiredLease} 가 next_attempt_at
+ * 을 "지금"으로 적는데 그 컬럼이 초 단위 DATETIME 이라, 반올림이 위로 튀느냐에 따라 같은
+ * 트랜잭션의 claim 이 집기도 하고 못 집기도 한다. 그래서 상태 대신 회수가 남긴 흔적으로
+ * 판정한다 — 회수만이 이 error_message 를 적고, 이어지는 claim 은 그것을 지우지 않는다.
+ */
+ @Test
+ void mergedPollStillRecoversAnExpiredLease() {
+ String deliveryId = uniqueId("merged-recover");
+ seedDelivery(deliveryId, WebhookDeliveryStatus.PROCESSING, null,
+ "worker-dead", LocalDateTime.now().minusMinutes(5));
+
+ webhookDeliveryRepository.recoverAndClaimPending("worker-alive", POLL_LIMIT);
+
+ Map row = fetchRow(deliveryId);
+ assertThat((String) row.get("error_message"))
+ .as("회수가 돌지 않았다면 이 문구는 적히지 않는다")
+ .isEqualTo("worker lease가 만료되어 webhook 처리를 다시 시도합니다.");
+ assertThat(row.get("lease_owner"))
+ .as("죽은 워커의 리스를 남겨두면 그 배달은 영영 처리되지 않는다")
+ .isIn(null, "worker-alive");
+ assertThat(row.get("status")).isIn(
+ WebhookDeliveryStatus.RETRY_WAIT.name(), WebhookDeliveryStatus.PROCESSING.name());
+ }
+
+ // ── 2. 깨우기의 트랜잭션 경계 ────────────────────────────────────────────────────────────────
+
+ @Test
+ void theWakeupSignalArrivesOnlyAfterTheEnqueueCommits() {
+ backOffFully();
+
+ transactionTemplate.executeWithoutResult(status -> {
+ eventPublisher.publishEvent(new WorkQueuedEvent(WorkQueue.WEBHOOK_DELIVERY));
+ assertThat(pollGate.shouldPoll(WorkQueue.WEBHOOK_DELIVERY))
+ .as("커밋 전에 깨우면 워커가 아직 보이지 않는 행을 찾다가 도로 물러난다")
+ .isFalse();
+ });
+
+ assertThat(pollGate.shouldPoll(WorkQueue.WEBHOOK_DELIVERY))
+ .as("커밋 직후에는 백오프가 얼마였든 다음 틱에 폴링해야 한다")
+ .isTrue();
+ }
+
+ @Test
+ void enqueueingARealDeliveryWakesTheWorkerQueue() {
+ backOffFully();
+
+ boolean accepted = webhookDeliveryRepository.enqueue(new WebhookDelivery(
+ uniqueId("merged-wake"), "push", "{}".getBytes(StandardCharsets.UTF_8)));
+
+ assertThat(accepted).isTrue();
+ assertThat(pollGate.shouldPoll(WorkQueue.WEBHOOK_DELIVERY))
+ .as("실제 enqueue 경로가 깨우기 신호를 내보내야 한다 — 아니면 배달이 백오프 상한만큼 방치된다")
+ .isTrue();
+ }
+
+ // ── helpers ──────────────────────────────────────────────────────────────────────────────
+
+ /**
+ * 게이트를 상한까지 물러나게 만든다. {@code recordIdle} 을 반복하면 간격이 두 배씩 늘어 상한에
+ * 닿는다 — 벽시계를 기다리지 않고 "충분히 물러난 상태"를 만드는 방법이다. 앞의
+ * {@code shouldPoll} 은 세대를 맞추기 위한 것으로, 이것이 없으면 이전 테스트가 남긴 깨우기
+ * 때문에 {@code recordIdle} 이 조용히 무시된다.
+ */
+ private void backOffFully() {
+ pollGate.shouldPoll(WorkQueue.WEBHOOK_DELIVERY);
+ for (int i = 0; i < 6; i++) {
+ pollGate.recordIdle(WorkQueue.WEBHOOK_DELIVERY);
+ }
+ assertThat(pollGate.shouldPoll(WorkQueue.WEBHOOK_DELIVERY)).isFalse();
+ }
+
+ private Callable> racePoll(CountDownLatch startBarrier, String workerId) {
+ return () -> {
+ startBarrier.countDown();
+ startBarrier.await();
+ return webhookDeliveryRepository.recoverAndClaimPending(workerId, POLL_LIMIT);
+ };
+ }
+
+ private String uniqueId(String label) {
+ String raw = "it-340-" + label + "-" + System.nanoTime();
+ return raw.length() <= 64 ? raw : raw.substring(0, 64);
+ }
+
+ private void seedDelivery(String deliveryId, WebhookDeliveryStatus status, LocalDateTime nextAttemptAt,
+ String leaseOwner, LocalDateTime leaseUntil) {
+ jdbcTemplate.update(
+ """
+ insert into webhook_deliveries
+ (delivery_id, event_type, payload, status, attempt, max_attempts,
+ next_attempt_at, lease_owner, lease_until)
+ values (?, ?, ?, ?, ?, ?, ?, ?, ?)
+ """,
+ deliveryId, "issues", "{}".getBytes(StandardCharsets.UTF_8), status.name(), 0, 5,
+ nextAttemptAt == null ? null : Timestamp.valueOf(nextAttemptAt),
+ leaseOwner,
+ leaseUntil == null ? null : Timestamp.valueOf(leaseUntil)
+ );
+ }
+
+ private Map fetchRow(String deliveryId) {
+ return jdbcTemplate.queryForMap("select * from webhook_deliveries where delivery_id = ?", deliveryId);
+ }
+}
diff --git a/src/test/java/com/example/dvely/webhook/infrastructure/worker/WebhookDeliveryWorkerTest.java b/src/test/java/com/example/dvely/webhook/infrastructure/worker/WebhookDeliveryWorkerTest.java
new file mode 100644
index 00000000..efc689a0
--- /dev/null
+++ b/src/test/java/com/example/dvely/webhook/infrastructure/worker/WebhookDeliveryWorkerTest.java
@@ -0,0 +1,110 @@
+package com.example.dvely.webhook.infrastructure.worker;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+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.example.dvely.common.worker.WorkerPollGate;
+import com.example.dvely.webhook.application.WebhookService;
+import com.example.dvely.webhook.domain.repository.WebhookDeliveryRepository;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.jupiter.api.Test;
+import org.springframework.core.task.TaskRejectedException;
+
+/**
+ * #340 5-3 — 웹훅 배달 처리를 스케줄러 스레드에서 떼어낸 뒤의 계약.
+ */
+class WebhookDeliveryWorkerTest {
+
+ private static final long BACKOFF_MS = 5_000L;
+
+ private final WebhookDeliveryRepository repository = mock(WebhookDeliveryRepository.class);
+ private final WebhookService webhookService = mock(WebhookService.class);
+
+ /** 백오프가 이 파일의 dispatch 검증에 끼어들지 않도록 게이트를 항상 열어 둔다. */
+ private static WorkerPollGate openGate() {
+ AtomicLong nanos = new AtomicLong();
+ return new WorkerPollGate(1000L, 30_000L, () -> nanos.addAndGet(TimeUnit.MINUTES.toNanos(1)));
+ }
+
+ private WebhookDeliveryWorker worker(Executor executor) {
+ return new WebhookDeliveryWorker(repository, webhookService, openGate(), executor, BACKOFF_MS);
+ }
+
+ @Test
+ void deliveriesAreHandedToTheExecutorInsteadOfRunningOnTheSchedulerThread() {
+ List submitted = new ArrayList<>();
+ WebhookDeliveryWorker worker = worker(submitted::add);
+ when(repository.recoverAndClaimPending(anyString(), eq(10)))
+ .thenReturn(List.of("delivery-1", "delivery-2"));
+
+ worker.dispatchPendingDeliveries();
+
+ // 스케줄러 스레드에서는 아무것도 처리하지 않는다 — 넘기기만 한다.
+ assertThat(submitted).hasSize(2);
+ verify(webhookService, never()).processDelivery(anyString());
+
+ submitted.forEach(Runnable::run);
+ verify(webhookService).processDelivery("delivery-1");
+ verify(webhookService).processDelivery("delivery-2");
+ }
+
+ /**
+ * 비동기로 넘기면 {@code TaskRejectedException} 이라는 새 실패 경로가 생긴다. 그것이 루프를
+ * 끊으면 같은 배치의 뒤 배달이 PROCESSING 인 채 리스 만료(2분)까지 남는다 — 배포·클라우드
+ * 워커에 있었던 것과 똑같은 버그를 여기서 새로 만들지 않는다.
+ */
+ @Test
+ void anExecutorRejectionReleasesThatClaimAndStillDeliversTheRest() {
+ List submitted = new ArrayList<>();
+ Executor rejectsFirst = new Executor() {
+ private boolean first = true;
+
+ @Override
+ public void execute(Runnable command) {
+ if (first) {
+ first = false;
+ throw new TaskRejectedException("webhookExecutor 포화");
+ }
+ submitted.add(command);
+ }
+ };
+ WebhookDeliveryWorker worker = worker(rejectsFirst);
+ when(repository.recoverAndClaimPending(anyString(), eq(10)))
+ .thenReturn(List.of("delivery-1", "delivery-2"));
+
+ worker.dispatchPendingDeliveries();
+
+ assertThat(submitted).hasSize(1);
+ verify(repository).releaseClaim(eq("delivery-1"), anyString(), eq(BACKOFF_MS));
+ verify(repository, never()).releaseClaim(eq("delivery-2"), anyString(), anyLong());
+ }
+
+ /** executor 스레드에서 튀어나온 예외가 조용히 사라지면 안 된다 — 워커가 잡아 남긴다. */
+ @Test
+ void anUnexpectedFailureInsideTheExecutorTaskDoesNotEscape() {
+ List submitted = new ArrayList<>();
+ WebhookDeliveryWorker worker = worker(submitted::add);
+ when(repository.recoverAndClaimPending(anyString(), eq(10))).thenReturn(List.of("delivery-1"));
+ doThrowOnProcess();
+
+ worker.dispatchPendingDeliveries();
+
+ assertThat(submitted).hasSize(1);
+ submitted.get(0).run(); // 예외가 밖으로 나오면 이 줄에서 테스트가 깨진다
+ }
+
+ private void doThrowOnProcess() {
+ org.mockito.Mockito.doThrow(new IllegalStateException("배달 행이 사라졌다"))
+ .when(webhookService).processDelivery("delivery-1");
+ }
+}