[Infra] 유휴 DB 쿼리 1904→74/분 · 배포·클라우드 워커의 행별 격리 누락 수정 [ #340 ]

src/main/java/com/example/dvely/agent/infrastructure/config/AsyncConfig.java

@@ -148,6 +148,51 @@ public Executor previewExecutor() {
         return executor;
     }
 
+    // #340 5-3: 웹훅 배달 처리를 스케줄러 스레드에서 떼어낸다. 핸들러가 GitHub API 를 호출하므로
+    // 한 배달이 느리면 그 동안 이 워커의 다음 폴링이 통째로 밀리고, 스케줄러 풀을 공유하는 다른
+    // 잡까지 굶는다. 한 폴링이 최대 CLAIM_BATCH_SIZE(10) 건을 넘기므로 큐를 그보다 넉넉히 둔다.
+    @Bean("webhookExecutor")
+    public ThreadPoolTaskExecutor webhookExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(2);
+        executor.setMaxPoolSize(4);
+        executor.setQueueCapacity(50);
+        executor.setThreadNamePrefix("webhook-");
+        executor.setAllowCoreThreadTimeOut(true);
+        executor.initialize();
+        return executor;
+    }
+
+    // #340 5-5: 도메인 검증 프로브(Cloudflare + GitHub + HTTPS)를 스케줄러 스레드에서 떼어낸다.
+    // 배치 20건을 직렬로 도는 동안 한 건이 타임아웃까지 버티면 그 시간이 그대로 스케줄러 점유가
+    // 된다. 동시 실행을 낮게 두는 이유는 이 호출들이 외부 API 레이트 리밋을 쓰기 때문이다.
+    @Bean("domainVerificationExecutor")
+    public ThreadPoolTaskExecutor domainVerificationExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(2);
+        executor.setMaxPoolSize(4);
+        executor.setQueueCapacity(50);
+        executor.setThreadNamePrefix("domain-verify-");
+        executor.setAllowCoreThreadTimeOut(true);
+        executor.initialize();
+        return executor;
+    }
+
+    // #340 5-10: 도커 prune 처럼 오래 걸리는 정비 작업 전용. prune 3회가 도커 데몬을 잠시 붙잡는
+    // 동안 스케줄러 스레드를 점유하지 않게 한다. 6시간에 한 번 도는 일이라 놀 때는 스레드를
+    // 회수한다(allowCoreThreadTimeOut).
+    @Bean("maintenanceExecutor")
+    public ThreadPoolTaskExecutor maintenanceExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(1);
+        executor.setMaxPoolSize(2);
+        executor.setQueueCapacity(10);
+        executor.setThreadNamePrefix("maintenance-");
+        executor.setAllowCoreThreadTimeOut(true);
+        executor.initialize();
+        return executor;
+    }
+
     @Bean("cloudConnectionExecutor")
     public Executor cloudConnectionExecutor() {
         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

src/main/java/com/example/dvely/agent/infrastructure/store/TaskStore.java

@@ -11,6 +11,8 @@
 import com.example.dvely.agent.infrastructure.persistence.entity.AgentRunEventEntity;
 import com.example.dvely.agent.infrastructure.persistence.repository.SpringDataAgentRunEventRepository;
 import com.example.dvely.agent.infrastructure.persistence.repository.SpringDataAgentRunRepository;
+import com.example.dvely.common.worker.WorkQueue;
+import com.example.dvely.common.worker.WorkQueuedEvent;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.time.Duration;
@@ -21,6 +23,7 @@
 import java.util.Optional;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Propagation;
@@ -52,6 +55,7 @@ public class TaskStore {
     private final SpringDataAgentRunRepository runRepository;
     private final SpringDataAgentRunEventRepository eventRepository;
     private final ObjectMapper objectMapper;
+    private final ApplicationEventPublisher eventPublisher;
 
     @Transactional
     public void save(AgentTask task) {
@@ -164,6 +168,7 @@ public void enqueue(String taskId) {
         AgentRunEntity run = requireRun(taskId);
         run.enqueue(false);
         appendEvent(taskId, "QUEUED", TaskStatus.QUEUED, "Agent task 실행을 대기합니다.");
+        signalWorkQueued();
     }
 
     @Transactional
@@ -177,9 +182,46 @@ public boolean retry(String taskId, Long ownerUserId) {
         }
         run.enqueue(true);
         appendEvent(taskId, "RETRY_QUEUED", TaskStatus.RETRY_WAIT, "수정안을 적용해 작업을 다시 실행합니다.");
+        signalWorkQueued();
         return true;
     }
 
+    /**
+     * 폴링 한 번이 하는 일 전부 — 만료 리스 회수와 claim 을 <b>한 트랜잭션</b>으로 묶는다(#340 5-1).
+     *
+     * <p>둘을 따로 부르면 폴링 한 번이 트랜잭션 두 개가 되고, 트랜잭션 하나는
+     * {@code SET autocommit=0} → 쿼리 → {@code COMMIT} → {@code SET autocommit=1} 로 왕복 네 번이다.
+     * 유휴 상태 실측에서 비용의 대부분이 SELECT 가 아니라 이 의례였다 — 합치는 것만으로 절반이
+     * 줄어든다.</p>
+     *
+     * <p>덤으로 회수 지연이 한 폴링 짧아진다. 회수 UPDATE 가 같은 트랜잭션에서 먼저 반영되므로,
+     * 방금 RETRY_WAIT 로 돌아온 태스크를 <b>같은 폴링의</b> claim 이 곧바로 집는다.</p>
+     *
+     * @param claimLimit 0 이하면 claim 쿼리를 아예 내지 않는다 — 실행기가 포화라 집어봐야 곧바로
+     *                   되돌려야 하는 상황(ADR-Y3)에서도 <b>회수는 계속 돌아야</b> 하기 때문이다.
+     *                   포화를 이유로 폴링을 통째로 건너뛰면 좀비 리스가 그만큼 오래 남는다.
+     */
+    @Transactional
+    public PollBatch recoverAndClaim(String workerId, int claimLimit) {
+        List<String> leaseExhausted = recoverExpiredLeases();
+        List<String> claimed = claimLimit > 0 ? claimRunnableTasks(workerId, claimLimit) : List.of();
+        return new PollBatch(leaseExhausted, claimed);
+    }
+
+    /**
+     * 폴링 한 번의 결과.
+     *
+     * @param leaseExhausted 복구 횟수를 소진해 FAILED 로 닫힌 taskId — 호출자가 사용자에게 알린다
+     * @param claimed        이번 폴링이 집은 taskId
+     */
+    public record PollBatch(List<String> leaseExhausted, List<String> claimed) {
+
+        /** 이번 폴링이 무언가라도 건드렸는가 — 워커의 백오프 판단 근거. */
+        public boolean touchedWork() {
+            return !leaseExhausted.isEmpty() || !claimed.isEmpty();
+        }
+    }
+
     @Transactional
     public List<String> claimRunnableTasks(String workerId, int limit) {
         List<String> candidates = runRepository.findRunnableTaskIds(
@@ -398,6 +440,7 @@ public void recoverStuckApproval(String taskId) {
                 TaskStatus.QUEUED,
                 "지연된 승인 처리를 복구해 작업을 시작합니다."
         );
+        signalWorkQueued();
     }
 
     /**
@@ -554,6 +597,7 @@ private boolean resumePastResultGate(String taskId, String eventType, String mes
             return false;
         }
         appendEvent(taskId, eventType, TaskStatus.QUEUED, message);
+        signalWorkQueued();
         return true;
     }
 
@@ -585,6 +629,19 @@ public void replacePlanAndRequeue(String taskId, AgentPlan newPlan) {
         }
         run.replacePlan(writePlan(newPlan));
         appendEvent(taskId, "REPLANNED", TaskStatus.QUEUED, "되묻기 답을 반영해 재계획했습니다.");
+        signalWorkQueued();
+    }
+
+    /**
+     * 이 태스크가 워커가 집을 수 있는 상태가 됐다고 알린다(#340 5-1).
+     *
+     * <p>{@code WorkerPollGate} 가 커밋 뒤에 받아 백오프를 즉시 푼다. 이 신호가 없으면, 유휴가
+     * 길어져 폴링 간격이 상한까지 늘어난 상태에서 사용자가 메시지를 보냈을 때 그 상한만큼
+     * 기다리게 된다 — 명백한 UX 회귀다. 신호가 있으면 백오프 값과 무관하게 다음 틱(≤1초)에
+     * 집힌다.</p>
+     */
+    private void signalWorkQueued() {
+        eventPublisher.publishEvent(new WorkQueuedEvent(WorkQueue.AGENT_RUN));
     }
 
     private String writeClarification(ClarificationRequest clarification) {
@@ -607,6 +664,7 @@ public boolean supplyInput(String taskId, Long ownerUserId, String value) {
         }
         run.supplyInput(value.trim());
         appendEvent(taskId, "INPUT_RECEIVED", TaskStatus.QUEUED, "사용자 입력을 받아 task를 다시 대기열에 넣었습니다.");
+        signalWorkQueued();
         return true;
     }
 

src/main/java/com/example/dvely/agent/infrastructure/worker/AgentRunWorker.java

@@ -6,6 +6,8 @@
 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.WorkerPollGate;
 import java.lang.management.ManagementFactory;
 import java.util.List;
 import java.util.Set;
@@ -37,6 +39,7 @@ public class AgentRunWorker {
     private final AgentMessageService agentMessageService;
     private final AgentExecutionRegistry executionRegistry;
     private final ThreadPoolTaskExecutor agentExecutor;
+    private final WorkerPollGate pollGate;
     private final long dispatchRejectBackoffMs;
     private final String workerId = ManagementFactory.getRuntimeMXBean().getName();
 
@@ -45,19 +48,25 @@ public AgentRunWorker(TaskStore taskStore,
                           AgentMessageService agentMessageService,
                           AgentExecutionRegistry executionRegistry,
                           @Qualifier("agentExecutor") ThreadPoolTaskExecutor agentExecutor,
+                          WorkerPollGate pollGate,
                           @Value("${qeploy.agent.worker.dispatch-reject-backoff-ms:5000}")
                           long dispatchRejectBackoffMs) {
         this.taskStore = taskStore;
         this.agentPlanExecutor = agentPlanExecutor;
         this.agentMessageService = agentMessageService;
         this.executionRegistry = executionRegistry;
         this.agentExecutor = agentExecutor;
+        this.pollGate = pollGate;
         this.dispatchRejectBackoffMs = dispatchRejectBackoffMs;
     }
 
     @Scheduled(fixedDelayString = "${qeploy.agent.worker.poll-interval-ms:1000}")
     public void dispatchQueuedRuns() {
-        notifyLeaseExhausted(taskStore.recoverExpiredLeases());
+        // #340 5-1: 틱은 여전히 1초마다 오지만, 일이 없는 동안에는 DB 를 치지 않는다. 게이트가
+        // 닫혀 있으면 여기서 끝이고 쿼리는 한 건도 나가지 않는다.
+        if (!pollGate.shouldPoll(WorkQueue.AGENT_RUN)) {
+            return;
+        }
 
         // ADR-Y3 SHOULD: best-effort capacity check before claiming at all. Deliberately racy (the
         // pool's real state can change the instant after this read) — it only needs to be
@@ -66,13 +75,32 @@ public void dispatchQueuedRuns() {
         // window where a claimed task shows as RUNNING while merely queued inside the executor
         // (audit §4.1's "상태 의미 왜곡" note).
         int freeSlots = estimateFreeExecutorSlots();
-        if (freeSlots <= 0) {
+        boolean saturated = freeSlots <= 0;
+        if (saturated) {
             log.debug("[AgentRunWorker] agentExecutor 포화로 이번 폴링은 claim을 생략합니다. workerId={}", workerId);
-            return;
         }
 
-        List<String> taskIds = taskStore.claimRunnableTasks(workerId, Math.min(CLAIM_BATCH_SIZE, freeSlots));
-        for (String taskId : taskIds) {
+        TaskStore.PollBatch batch;
+        try {
+            // 포화여도 회수는 돌린다(claimLimit=0). 포화를 이유로 폴링을 통째로 건너뛰면 좀비
+            // 리스가 그만큼 오래 남는다.
+            batch = taskStore.recoverAndClaim(workerId, saturated ? 0 : Math.min(CLAIM_BATCH_SIZE, freeSlots));
+        } catch (RuntimeException exception) {
+            // DB 가 흔들리는 동안 매초 같은 쿼리를 다시 던져봐야 소용이 없다 — 물러나며 재시도한다.
+            pollGate.recordIdle(WorkQueue.AGENT_RUN);
+            throw exception;
+        }
+
+        // 포화로 claim 을 생략한 것은 "일이 없다"가 아니다. 자리가 나는 것을 알려줄 신호는 없으므로
+        // 여기서 물러나면 큐에 쌓인 태스크가 백오프 상한만큼 늦게 출발한다.
+        if (saturated || batch.touchedWork()) {
+            pollGate.recordBusy(WorkQueue.AGENT_RUN);
+        } else {
+            pollGate.recordIdle(WorkQueue.AGENT_RUN);
+        }
+
+        notifyLeaseExhausted(batch.leaseExhausted());
+        for (String taskId : batch.claimed()) {
             dispatchOne(taskId);
         }
     }

src/main/java/com/example/dvely/agent/infrastructure/worker/DockerGarbageSweeper.java

@@ -2,8 +2,9 @@
 
 import com.example.dvely.agent.infrastructure.docker.DockerContainerService;
 import java.time.Duration;
-import lombok.RequiredArgsConstructor;
+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;
@@ -30,10 +31,16 @@
  */
 @Slf4j
 @Component
-@RequiredArgsConstructor
 public class DockerGarbageSweeper {
 
     private final DockerContainerService dockerService;
+    private final Executor maintenanceExecutor;
+
+    public DockerGarbageSweeper(DockerContainerService dockerService,
+                                @Qualifier("maintenanceExecutor") Executor maintenanceExecutor) {
+        this.dockerService = dockerService;
+        this.maintenanceExecutor = maintenanceExecutor;
+    }
 
     /** 이 시간 안에 쓰인 빌드 캐시는 남긴다. */
     @Value("${qeploy.docker.sweep.build-cache-keep-hours:48}")
@@ -53,6 +60,12 @@ public void sweep() {
         if (!enabled) {
             return;
         }
+        // #340 5-10: prune 3회는 도커 데몬을 잠시 붙잡는다. 그 시간을 스케줄러 스레드로 때우면
+        // 같은 풀을 쓰는 다른 잡이 그만큼 밀린다 — 6시간에 한 번이라도 전용 스레드로 넘긴다.
+        maintenanceExecutor.execute(this::pruneGarbage);
+    }
+
+    private void pruneGarbage() {
         long freed = dockerService.pruneGarbage(Duration.ofHours(buildCacheKeepHours));
         if (freed < 0) {
             return;   // 도커에 못 닿음 — pruneGarbage 가 이미 경고를 남겼다

src/main/java/com/example/dvely/chat/application/command/ChatCommandService.java

@@ -96,16 +96,15 @@ public void deleteConversationsForProject(Long userId, Long projectId) {
         }
     }
 
+    /**
+     * 만료된 휴지통 대화를 영구 삭제한다.
+     *
+     * <p>#340 5-9: 엔티티를 전부 로드한 뒤 {@code deleteById} 를 N 번 부르던 것을 벌크 DELETE
+     * 한 문장으로 바꿨다. 지우려고 읽을 이유가 없다 — 삭제 조건이 곧 SELECT 조건이었다.</p>
+     */
     @Transactional
     public int purgeExpiredConversations() {
-        List<Conversation> expired = conversationRepository.findAllByDeletedTrueAndDeletedAtLessThanEqual(
-                ChatTrashPolicy.cutoff(LocalDateTime.now())
-        );
-        expired.stream()
-                .map(Conversation::getId)
-                .filter(java.util.Objects::nonNull)
-                .forEach(conversationRepository::deleteById);
-        return expired.size();
+        return conversationRepository.deleteExpiredTrash(ChatTrashPolicy.cutoff(LocalDateTime.now()));
     }
 
     @Transactional

src/main/java/com/example/dvely/chat/domain/repository/ConversationRepository.java

@@ -21,5 +21,12 @@ public interface ConversationRepository {
 
     void deleteById(Long conversationId);
 
+    /**
+     * 만료된 휴지통 대화를 한 문장으로 지운다(#340 5-9).
+     *
+     * @return 지워진 대화 수
+     */
+    int deleteExpiredTrash(LocalDateTime cutoff);
+
     Conversation save(Conversation conversation);
 }

src/main/java/com/example/dvely/chat/infrastructure/persistence/repository/ConversationRepositoryAdapter.java

@@ -60,6 +60,11 @@ public void deleteById(Long conversationId) {
         springDataConversationRepository.deleteById(conversationId);
     }
 
+    @Override
+    public int deleteExpiredTrash(LocalDateTime cutoff) {
+        return springDataConversationRepository.deleteExpiredTrash(cutoff);
+    }
+
     @Override
     public Conversation save(Conversation conversation) {
         ConversationEntity entity;

src/main/java/com/example/dvely/chat/infrastructure/persistence/repository/SpringDataConversationRepository.java

@@ -5,6 +5,9 @@
 import java.util.List;
 import java.util.Optional;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
 
 public interface SpringDataConversationRepository extends JpaRepository<ConversationEntity, Long> {
 
@@ -19,4 +22,16 @@ public interface SpringDataConversationRepository extends JpaRepository<Conversa
     Optional<ConversationEntity> findByIdAndUserIdAndDeletedFalse(Long conversationId, Long userId);
 
     Optional<ConversationEntity> findByIdAndUserId(Long conversationId, Long userId);
+
+    // #340 5-9: 만료된 휴지통 대화를 한 문장으로 지운다. 예전에는 엔티티를 전부 로드한 뒤
+    // deleteById 를 N 번 불렀다 — 지우려고 읽고, 지우려고 또 한 번씩 왕복했다. 파생
+    // deleteBy... 메서드도 내부적으로 같은 짓을 하므로 명시적 벌크 DELETE 로 적는다.
+    // 여기서 지우는 대상은 이미 소프트 삭제돼 사용자 화면에서 사라진 대화뿐이다.
+    @Modifying(clearAutomatically = true, flushAutomatically = true)
+    @Query("""
+            delete from ConversationEntity conversation
+             where conversation.deleted = true
+               and conversation.deletedAt <= :cutoff
+            """)
+    int deleteExpiredTrash(@Param("cutoff") LocalDateTime cutoff);
 }

src/main/java/com/example/dvely/cloudconnection/domain/repository/CloudConnectionVerificationJobRepository.java

@@ -16,5 +16,21 @@ public interface CloudConnectionVerificationJobRepository {
 
     List<String> claimPending(String workerId, int limit);
 
+    /**
+     * 폴링 한 번이 하는 일 전부 — 만료 리스 회수와 claim 을 <b>한 트랜잭션</b>으로 묶는다(#340 5-1).
+     * 따로 부르면 폴링 한 번이 트랜잭션 두 개가 되고, 트랜잭션마다 붙는 {@code SET autocommit} ·
+     * {@code COMMIT} 의례가 유휴 DB 비용의 대부분이었다. 회수 UPDATE 가 같은 트랜잭션에서 먼저
+     * 반영되므로, 방금 회수된 행을 같은 폴링의 claim 이 곧바로 집는다.
+     */
+    List<String> recoverAndClaimPending(String workerId, int limit);
+
     void recoverExpiredLeases();
+
+    /**
+     * claim 해 놓고 실행기에 넘기지 못한 job 을 PENDING 으로 되돌린다(#340 5-2).
+     *
+     * @return 이 호출이 실제로 되돌렸으면 true. false 는 그 사이 다른 주체가 이미 이 행을
+     *         RUNNING 밖으로 옮겼다는 뜻이다.
+     */
+    boolean releaseClaim(String jobId, String workerId);
 }