From 56c7d89f5ed38af609576be95ee62504875a28cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=B0=95=EC=9E=AC=ED=99=8D?= <105151063+PHJ2000@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:43:44 +0900 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20=EC=9D=B8=EC=A6=9D=20=EC=9D=B4?= =?UTF-8?q?=EB=A9=94=EC=9D=BC=20=EC=9E=91=EC=97=85=20=EC=98=81=EC=86=8D?= =?UTF-8?q?=ED=99=94=20=EB=B0=8F=20=EC=9E=A5=EC=95=A0=20=EB=B3=B5=EA=B5=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 토큰과 이메일 작업을 원자적으로 저장하고 DB 선점·시도 이력으로 복구 - 최대 4회 지수 백오프와 SMTP 오류 분류, 오래된 선점자의 완료 갱신 차단 - 인증 링크 암호화와 최종 payload 삭제, 중복 발송 한계 및 배포 절차 명시 - 재시작·동시 실행·실제 DB 기록 실패·쿨다운 회귀 테스트 추가 검증: 전체 324 tests, assemble, Checkstyle main/test, Spotless, UTF-8, Gitleaks 통과 Related #270 --- README.md | 36 ++ deploy/backend-ec2.env.example | 6 + .../auth/repository/EmailJobStore.java | 130 +++++ .../auth/service/AsyncEmailService.java | 81 --- .../auth/service/EmailEventListener.java | 10 +- .../auth/service/EmailJobScheduler.java | 34 ++ .../auth/service/EmailJobService.java | 38 ++ .../auth/service/EmailJobWorker.java | 105 ++++ .../auth/service/EmailPayloadCipher.java | 57 ++ .../service/EmailVerificationService.java | 7 +- .../service/RetryingEmailDeliveryService.java | 102 ---- .../auth/service/SmtpFailureClassifier.java | 44 ++ src/main/resources/application.yml | 9 + .../db/migration/V15__create_email_jobs.sql | 24 + .../AsyncEmailRetryIntegrationTest.java | 485 ------------------ .../integration/EmailJobIntegrationTest.java | 350 +++++++++++++ ...va => EmailJobTracingIntegrationTest.java} | 123 ++--- ...> EmailJobTransactionIntegrationTest.java} | 46 +- .../auth/service/AsyncEmailServiceTest.java | 78 --- .../auth/service/EmailEventListenerTest.java | 6 +- .../auth/service/EmailPayloadCipherTest.java | 21 + .../service/EmailVerificationServiceTest.java | 2 - .../RetryingEmailDeliveryServiceTest.java | 62 --- .../service/SmtpFailureClassifierTest.java | 59 +++ src/test/resources/application-test.yml | 4 + 25 files changed, 1003 insertions(+), 916 deletions(-) create mode 100644 src/main/java/com/beachcheck/auth/repository/EmailJobStore.java delete mode 100644 src/main/java/com/beachcheck/auth/service/AsyncEmailService.java create mode 100644 src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java create mode 100644 src/main/java/com/beachcheck/auth/service/EmailJobService.java create mode 100644 src/main/java/com/beachcheck/auth/service/EmailJobWorker.java create mode 100644 src/main/java/com/beachcheck/auth/service/EmailPayloadCipher.java delete mode 100644 src/main/java/com/beachcheck/auth/service/RetryingEmailDeliveryService.java create mode 100644 src/main/java/com/beachcheck/auth/service/SmtpFailureClassifier.java create mode 100644 src/main/resources/db/migration/V15__create_email_jobs.sql delete mode 100644 src/test/java/com/beachcheck/auth/integration/AsyncEmailRetryIntegrationTest.java create mode 100644 src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java rename src/test/java/com/beachcheck/auth/integration/{AsyncEmailTracingIntegrationTest.java => EmailJobTracingIntegrationTest.java} (71%) rename src/test/java/com/beachcheck/auth/integration/{EmailAfterCommitIntegrationTest.java => EmailJobTransactionIntegrationTest.java} (70%) delete mode 100644 src/test/java/com/beachcheck/auth/service/AsyncEmailServiceTest.java create mode 100644 src/test/java/com/beachcheck/auth/service/EmailPayloadCipherTest.java delete mode 100644 src/test/java/com/beachcheck/auth/service/RetryingEmailDeliveryServiceTest.java create mode 100644 src/test/java/com/beachcheck/auth/service/SmtpFailureClassifierTest.java diff --git a/README.md b/README.md index 4375c23d..34903be1 100644 --- a/README.md +++ b/README.md @@ -207,3 +207,39 @@ Beach_complex/ ## 라이선스 [MIT License](./LICENSE) + +### 인증 이메일 작업 복구 + +인증 토큰과 `email_jobs`는 같은 DB 트랜잭션에서 저장한다. 워커는 시작 시와 기본 1초 주기로 +`PENDING`, 재시도 시각이 지난 `RETRY_WAIT`, lease가 지난 `SENDING`을 조회한다. +`FOR UPDATE SKIP LOCKED`로 선점·시도 횟수 증가·시도 이력 생성을 커밋한 다음 SMTP를 호출한다. +완료 기록은 현재 시도 번호가 일치할 때만 허용한다. 상태는 +`PENDING → SENDING → SUCCEEDED / RETRY_WAIT / FAILED`이며 `email_job_attempts`에 각 시도를 남긴다. + +최대 4회는 최초 시도를 포함하며, 재시도 간격은 5→10→20초다. 네트워크/SMTP 4xx 오류는 재시도하고, +SMTP 5xx·인증·주소/메시지 오류는 즉시 실패한다. 코드를 알 수 없는 `MailSendException`은 한도 내에서 +재시도한다. 자동 재시도는 같은 토큰/작업을 사용해 3분 수동 재전송 제한을 갱신하거나 우회하지 않는다. +만료·사용·재전송으로 무효화된 토큰의 대기 작업은 발송 전에 `FAILED`로 종료한다. +선점 이후 토큰이 무효화되는 경쟁에서는 이미 시작한 SMTP 전송을 취소할 수 없다. + +SMTP가 수락한 뒤 DB 성공 기록 전에 종료되면 발송 여부를 확정할 수 없다. +해당 시도는 복구 시 `UNKNOWN`으로 남고 다시 보낼 수 있으므로 **exactly-once를 보장하지 않는다**. +유효한 lease 안에서는 한 워커만 작업을 가져가지만, SMTP 지연이나 JVM 중단이 lease를 넘으면 +이전 워커와 복구 워커의 전송이 겹칠 수도 있다. 작업 ID·시도 이력·완료 조건부 갱신은 중복을 줄이고 +조사를 돕지만 수신자의 중복 수신을 제거하지 못한다. `SUCCEEDED`는 SMTP 수락이며 받은편지함 도착이 아니다. +4번째 시도가 불명 상태로 끝나면 추가 발송 없이 `FAILED`/`ATTEMPTS_EXHAUSTED_UNKNOWN`으로 남긴다. + +배포 전 모든 인스턴스에 `APP_EMAIL_JOBS_PAYLOAD_KEY`를 동일하게 설정해야 한다. +`openssl rand -base64 32`로 생성한 32바이트 키를 Secret으로 보관하며 미설정/잘못된 길이는 시작을 차단한다. +인증 링크는 AES-GCM으로 암호화해 저장하고 종료된 작업에서는 삭제한다. 키를 변경하기 전에는 +미완료 작업을 처리하거나 기존 키로 복호화 후 새 키로 재암호화해야 한다. 수신 주소와 시도 이력은 DB에 +남으므로 접근을 제한하며, 이 변경에는 자동 보존기간 삭제 작업을 포함하지 않는다. +주소·링크·제목·본문·SMTP 예외 원문은 로그/Trace에 기록하지 않는다. +작업 ID는 로그와 DB에만 남기고, 각 시도의 새 Trace는 저장된 생산 Trace에 Span Link로 연결한다. + +설정은 [배포 환경 예시](deploy/backend-ec2.env.example)를 참고한다. 기본 lease는 300초(최소 60초), +SMTP 연결/읽기/쓰기 timeout은 각각 5초다. timeout은 전체 전송 시간의 엄밀한 상한은 아니다. +`APP_EMAIL_JOBS_POLLING_ENABLED=false`는 시작 복구와 주기 실행을 모두 끄며 이미 진행 중인 전송을 +취소하지 않는다. 롤백 전 워커를 정지하고 진행 중 요청을 종료한 뒤 미완료 작업을 확인한다. +구버전은 DB 작업을 처리하지 못하며, 기존 메모리 작업을 새 테이블로 자동 이전하지 않는다. +배포 전 구버전 요청을 drain해야 하며 운영 배포와 실제 SMTP 발송 검증은 PR 테스트 범위와 별개다. diff --git a/deploy/backend-ec2.env.example b/deploy/backend-ec2.env.example index 4938f072..86bde2e7 100644 --- a/deploy/backend-ec2.env.example +++ b/deploy/backend-ec2.env.example @@ -13,3 +13,9 @@ APP_EMAIL_VERIFICATION_BASE_URL=https:///api/auth/verify-email SPRING_MAIL_USERNAME= SPRING_MAIL_PASSWORD= APP_EMAIL_FROM_ADDRESS= +# Required: generate once with `openssl rand -base64 32`; same key on every email worker. +# Preserve this secret across restarts; drain outstanding jobs before rotation. +APP_EMAIL_JOBS_PAYLOAD_KEY= +APP_EMAIL_JOBS_POLLING_ENABLED=true +APP_EMAIL_JOBS_POLL_INTERVAL_MS=1000 +APP_EMAIL_JOBS_LEASE_SECONDS=300 diff --git a/src/main/java/com/beachcheck/auth/repository/EmailJobStore.java b/src/main/java/com/beachcheck/auth/repository/EmailJobStore.java new file mode 100644 index 00000000..31b8a34e --- /dev/null +++ b/src/main/java/com/beachcheck/auth/repository/EmailJobStore.java @@ -0,0 +1,130 @@ +package com.beachcheck.auth.repository; + +import java.util.Optional; +import java.util.UUID; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.util.Assert; + +@Repository +public class EmailJobStore { + private final JdbcTemplate jdbc; + private final long leaseSeconds; + + public EmailJobStore( + JdbcTemplate jdbc, @Value("${app.email.jobs.lease-seconds:300}") long leaseSeconds) { + Assert.isTrue(leaseSeconds >= 60, "Email lease must be at least 60 seconds"); + this.jdbc = jdbc; + this.leaseSeconds = leaseSeconds; + } + + @Transactional(propagation = Propagation.MANDATORY) + public UUID enqueue(String recipient, String payload, String tokenHash, String traceparent) { + UUID id = UUID.randomUUID(); + jdbc.update( + "INSERT INTO email_jobs(id, recipient, payload, token_hash, producer_traceparent) VALUES (?, ?, ?, ?, ?)", + id, + recipient, + payload, + tokenHash, + traceparent); + return id; + } + + /** 선점·횟수 증가·이력 생성은 한 트랜잭션이며 SMTP 호출 전에 커밋한다. */ + @Transactional(propagation = Propagation.REQUIRES_NEW) + public Optional claim() { + var candidates = + jdbc.query( + """ + SELECT id, recipient, payload, attempts, producer_traceparent, + EXISTS (SELECT 1 FROM email_verification_tokens t WHERE t.token = token_hash + AND t.used_at IS NULL AND t.expires_at > CURRENT_TIMESTAMP) AS valid_token + FROM email_jobs + WHERE status = 'PENDING' + OR (status = 'RETRY_WAIT' AND next_retry_at <= CURRENT_TIMESTAMP) + OR (status = 'SENDING' AND lease_until <= CURRENT_TIMESTAMP) + ORDER BY created_at, id LIMIT 1 FOR UPDATE SKIP LOCKED + """, + (rs, row) -> + new Candidate( + new Job( + rs.getObject("id", UUID.class), + rs.getString("recipient"), + rs.getString("payload"), + rs.getInt("attempts"), + rs.getString("producer_traceparent")), + rs.getBoolean("valid_token"))); + if (candidates.isEmpty()) return Optional.empty(); + Candidate candidate = candidates.getFirst(); + Job old = candidate.job(); + jdbc.update( + """ + UPDATE email_job_attempts SET outcome = 'UNKNOWN', finished_at = CURRENT_TIMESTAMP + WHERE job_id = ? AND outcome = 'STARTED' + """, + old.id()); + if (old.attempt() >= 4 || !candidate.validToken()) { + jdbc.update( + """ + UPDATE email_jobs SET status = 'FAILED', payload = NULL, lease_until = NULL, + next_retry_at = NULL, last_error = ? WHERE id = ? + """, + old.attempt() >= 4 ? "ATTEMPTS_EXHAUSTED_UNKNOWN" : "TOKEN_INACTIVE", + old.id()); + // 다음 poll에서 다음 후보를 조회하며, 실패 행은 다시 선택하지 않는다. + return Optional.empty(); + } + int attempt = old.attempt() + 1; + jdbc.update( + """ + UPDATE email_jobs SET status = 'SENDING', attempts = ?, last_attempt_at = CURRENT_TIMESTAMP, + next_retry_at = NULL, lease_until = CURRENT_TIMESTAMP + (? * INTERVAL '1 second') + WHERE id = ? + """, + attempt, + leaseSeconds, + old.id()); + jdbc.update("INSERT INTO email_job_attempts(job_id, attempt) VALUES (?, ?)", old.id(), attempt); + return Optional.of( + new Job(old.id(), old.recipient(), old.payload(), attempt, old.traceparent())); + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public boolean finish(Job job, String status, String outcome, long delayMs) { + int changed = + jdbc.update( + """ + UPDATE email_jobs SET status = ?, lease_until = NULL, + next_retry_at = CASE WHEN ? = 'RETRY_WAIT' THEN CURRENT_TIMESTAMP + (? * INTERVAL '1 millisecond') ELSE NULL END, + payload = CASE WHEN ? IN ('SUCCEEDED', 'FAILED') THEN NULL ELSE payload END, + last_error = CASE WHEN ? = 'SUCCEEDED' THEN NULL ELSE ? END + WHERE id = ? AND status = 'SENDING' AND attempts = ? + """, + status, + status, + delayMs, + status, + status, + outcome, + job.id(), + job.attempt()); + if (changed == 0) return false; + jdbc.update( + """ + UPDATE email_job_attempts SET outcome = ?, finished_at = CURRENT_TIMESTAMP + WHERE job_id = ? AND attempt = ? + """, + outcome, + job.id(), + job.attempt()); + return true; + } + + public record Job(UUID id, String recipient, String payload, int attempt, String traceparent) {} + + private record Candidate(Job job, boolean validToken) {} +} diff --git a/src/main/java/com/beachcheck/auth/service/AsyncEmailService.java b/src/main/java/com/beachcheck/auth/service/AsyncEmailService.java deleted file mode 100644 index b561517d..00000000 --- a/src/main/java/com/beachcheck/auth/service/AsyncEmailService.java +++ /dev/null @@ -1,81 +0,0 @@ -package com.beachcheck.auth.service; - -import io.micrometer.tracing.Span; -import io.micrometer.tracing.Tracer; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.scheduling.annotation.Async; -import org.springframework.stereotype.Service; - -/** - * Why: 이메일 발송을 비동기로 처리하며 SMTP 장애 시 자동 재시도를 수행한다. - * - *

Policy: 기술 계층의 장애 복구 메커니즘으로, MailSendException 발생 시 지수 백오프 전략으로 최대 3회 자동 재시도한다. - * - *

Contract(Input/Output): 이메일 발송 메서드를 제공한다. - * - *

Note: EmailVerificationService에서 분리한 이유는 @Async가 같은 클래스 내부 호출에서는 작동하지 않기 때문이다. 비동기 작업 span이 - * SMTP 재시도 전체를 한 번만 감싸도록 재시도 경계는 {@link RetryingEmailDeliveryService}에 위임한다. - */ -@Service -public class AsyncEmailService { - - private static final String EMAIL_OPERATION = "verification"; - - private final RetryingEmailDeliveryService emailDeliveryService; - private final Tracer tracer; - private final String fromAddress; - private final long tokenExpirationMinutes; - - public AsyncEmailService( - RetryingEmailDeliveryService emailDeliveryService, - Tracer tracer, - @Value("${app.email-verification.from-address:}") String fromAddress, - @Value("${app.email-verification.token-expiration-minutes:30}") long tokenExpirationMinutes) { - this.emailDeliveryService = emailDeliveryService; - this.tracer = tracer; - this.fromAddress = fromAddress; - this.tokenExpirationMinutes = tokenExpirationMinutes; - } - - /** - * Why: 이메일 인증 메일을 비동기로 발송하며, SMTP 서버 장애 시 자동 재시도를 수행한다. - * - *

Policy: 기술 계층의 장애 복구 메커니즘으로, MailSendException 발생 시 지수 백오프 전략으로 최대 3회 자동 재시도한다. (5초 → 10초 → - * 20초 대기). 회원가입 API는 즉시 응답하며, 이메일 발송은 백그라운드에서 처리된다. maxAttempts=4는 최초 시도 1회 + 재시도 3회를 의미한다. - * - *

Contract(Input): to는 수신자 이메일 주소, verificationLink는 인증 링크 전체 URL이다. - * - *

Contract(Output): 이메일 작업 전체를 {@code email.verification.send} span 하나로 기록한다. - * - * @see org.springframework.scheduling.annotation.Async - * @see RetryingEmailDeliveryService - */ - @Async("emailTaskExecutor") - public void sendVerificationEmailAsync(String to, String verificationLink) { - - String subject = "이메일 인증"; - String body = - """ - 아래 링크를 클릭하여 이메일을 인증해주세요: - - %s - - 이 링크는 %d분 후에 만료됩니다. - """ - .formatted(verificationLink, tokenExpirationMinutes); - - Span emailSpan = - tracer - .nextSpan() - .name("email.verification.send") - .tag("email.operation", EMAIL_OPERATION) - .start(); - try (Tracer.SpanInScope ignored = tracer.withSpan(emailSpan)) { - RetryingEmailDeliveryService.DeliveryOutcome outcome = - emailDeliveryService.sendVerificationEmail(fromAddress, to, subject, body); - emailSpan.tag("email.delivery.outcome", outcome.value()); - } finally { - emailSpan.end(); - } - } -} diff --git a/src/main/java/com/beachcheck/auth/service/EmailEventListener.java b/src/main/java/com/beachcheck/auth/service/EmailEventListener.java index 3983c3f5..3b94b0be 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailEventListener.java +++ b/src/main/java/com/beachcheck/auth/service/EmailEventListener.java @@ -7,14 +7,14 @@ @Component public class EmailEventListener { - private final AsyncEmailService asyncEmailService; + private final EmailJobService emailJobService; - public EmailEventListener(AsyncEmailService asyncEmailService) { - this.asyncEmailService = asyncEmailService; + public EmailEventListener(EmailJobService emailJobService) { + this.emailJobService = emailJobService; } - @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT) public void handleEmailVerificationEvent(EmailVerificationEvent event) { - asyncEmailService.sendVerificationEmailAsync(event.email(), event.verificationLink()); + emailJobService.enqueue(event.email(), event.verificationLink()); } } diff --git a/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java b/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java new file mode 100644 index 00000000..76c06cf4 --- /dev/null +++ b/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java @@ -0,0 +1,34 @@ +package com.beachcheck.auth.service; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +@Component +@ConditionalOnProperty( + name = "app.email.jobs.polling-enabled", + havingValue = "true", + matchIfMissing = true) +public class EmailJobScheduler { + private static final Logger log = LoggerFactory.getLogger(EmailJobScheduler.class); + private final EmailJobWorker worker; + + public EmailJobScheduler(EmailJobWorker worker) { + this.worker = worker; + } + + @EventListener(ApplicationReadyEvent.class) + @Scheduled(fixedDelayString = "${app.email.jobs.poll-interval-ms:1000}") + public void poll() { + try { + worker.poll(); + } catch (RuntimeException failure) { + // DB/SMTP 예외 원문은 주소·인증 링크를 포함할 수 있다. + log.error("Email job polling failed: errorType={}", failure.getClass().getName()); + } + } +} diff --git a/src/main/java/com/beachcheck/auth/service/EmailJobService.java b/src/main/java/com/beachcheck/auth/service/EmailJobService.java new file mode 100644 index 00000000..311d390b --- /dev/null +++ b/src/main/java/com/beachcheck/auth/service/EmailJobService.java @@ -0,0 +1,38 @@ +package com.beachcheck.auth.service; + +import com.beachcheck.auth.repository.EmailJobStore; +import com.beachcheck.global.util.HashUtils; +import io.micrometer.tracing.Span; +import io.micrometer.tracing.Tracer; +import java.util.UUID; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +@Service +public class EmailJobService { + private final EmailJobStore store; + private final EmailPayloadCipher cipher; + private final Tracer tracer; + + public EmailJobService(EmailJobStore store, EmailPayloadCipher cipher, Tracer tracer) { + this.store = store; + this.cipher = cipher; + this.tracer = tracer; + } + + @Transactional + public UUID enqueue(String email, String verificationLink) { + String rawToken = verificationLink.substring(verificationLink.lastIndexOf("?token=") + 7); + Span current = tracer.currentSpan(); + String parent = + current == null + ? null + : "00-" + + current.context().traceId() + + "-" + + current.context().spanId() + + (Boolean.TRUE.equals(current.context().sampled()) ? "-01" : "-00"); + return store.enqueue( + email, cipher.encrypt(verificationLink), HashUtils.sha256Hex(rawToken), parent); + } +} diff --git a/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java b/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java new file mode 100644 index 00000000..4b1a3d91 --- /dev/null +++ b/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java @@ -0,0 +1,105 @@ +package com.beachcheck.auth.service; + +import com.beachcheck.auth.repository.EmailJobStore; +import com.beachcheck.auth.repository.EmailJobStore.Job; +import com.beachcheck.global.util.HashUtils; +import io.micrometer.tracing.Link; +import io.micrometer.tracing.Span; +import io.micrometer.tracing.Tracer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +@Service +public class EmailJobWorker { + private static final Logger log = LoggerFactory.getLogger(EmailJobWorker.class); + private final EmailJobStore store; + private final EmailSender sender; + private final EmailPayloadCipher cipher; + private final SmtpFailureClassifier classifier; + private final Tracer tracer; + private final String from; + + public EmailJobWorker( + EmailJobStore store, + EmailSender sender, + EmailPayloadCipher cipher, + SmtpFailureClassifier classifier, + Tracer tracer, + @Value("${app.email-verification.from-address:}") String from) { + this.store = store; + this.sender = sender; + this.cipher = cipher; + this.classifier = classifier; + this.tracer = tracer; + this.from = from; + } + + public void poll() { + // 직렬로 소량 처리하여 메모리 큐에 선점한 작업을 오래 보관하지 않는다. + for (int i = 0; i < 20; i++) { + var claimed = store.claim(); + if (claimed.isEmpty()) return; + deliver(claimed.get()); + } + } + + private void deliver(Job job) { + Span.Builder builder = + tracer.spanBuilder().setNoParent().name("email.verification.send").kind(Span.Kind.CONSUMER); + String parent = job.traceparent(); + if (parent != null && parent.matches("00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]")) { + String[] parts = parent.split("-"); + builder.addLink( + new Link( + tracer + .traceContextBuilder() + .traceId(parts[1]) + .spanId(parts[2]) + .sampled(parts[3].equals("01")) + .build())); + } + Span work = builder.start().tag("email.operation", "verification"); + try (Tracer.SpanInScope ignored = tracer.withSpan(work)) { + String status = "SUCCEEDED"; + String outcome = "ACCEPTED"; + Span smtp = + tracer + .nextSpan() + .name("email.smtp.send") + .tag("email.retry.attempt", String.valueOf(job.attempt())) + .start(); + try (Tracer.SpanInScope smtpScope = tracer.withSpan(smtp)) { + sender.send( + from, + job.recipient(), + "이메일 인증", + "아래 링크를 클릭하여 이메일을 인증해주세요:\n\n" + cipher.decrypt(job.payload())); + smtp.tag("email.delivery.outcome", "success"); + } catch (RuntimeException failure) { + boolean retry = classifier.retryable(failure); + status = retry && job.attempt() < 4 ? "RETRY_WAIT" : "FAILED"; + outcome = retry ? "TRANSIENT_SMTP_FAILURE" : "PERMANENT_FAILURE"; + smtp.tag("email.delivery.outcome", "error"); + smtp.tag("error.type", failure.getClass().getName()); + smtp.error(new IllegalStateException("SMTP 이메일 전송 실패")); + } finally { + smtp.end(); + } + // 성공 기록 실패는 SMTP 실패로 분류하지 않는다. SENDING을 보존해 lease 만료 후 복구한다. + boolean recorded = store.finish(job, status, outcome, 5000L << (job.attempt() - 1)); + work.tag( + "email.delivery.outcome", + recorded ? status.toLowerCase(java.util.Locale.ROOT) : "stale_claim"); + log.info( + "Email job={} attempt={} outcome={} recipientHash={}", + job.id(), + job.attempt(), + recorded ? status : "STALE_CLAIM", + HashUtils.sha256Hex(job.recipient())); + } finally { + work.end(); + } + } +} diff --git a/src/main/java/com/beachcheck/auth/service/EmailPayloadCipher.java b/src/main/java/com/beachcheck/auth/service/EmailPayloadCipher.java new file mode 100644 index 00000000..c8d88ae8 --- /dev/null +++ b/src/main/java/com/beachcheck/auth/service/EmailPayloadCipher.java @@ -0,0 +1,57 @@ +package com.beachcheck.auth.service; + +import java.nio.charset.StandardCharsets; +import java.security.GeneralSecurityException; +import java.security.SecureRandom; +import java.util.Base64; +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +/** 인증 링크 원문을 DB에 노출하지 않는다. 키는 모든 워커에서 동일하게 유지한다. */ +@Component +public class EmailPayloadCipher { + private final byte[] key; + private final SecureRandom random = new SecureRandom(); + + public EmailPayloadCipher(@Value("${app.email.jobs.payload-key:}") String encodedKey) { + if (encodedKey.isBlank()) { + throw new IllegalArgumentException("APP_EMAIL_JOBS_PAYLOAD_KEY is required for email jobs"); + } + key = Base64.getDecoder().decode(encodedKey); + if (key.length != 32) { + throw new IllegalArgumentException("Email payload key must contain 32 bytes"); + } + } + + public String encrypt(String text) { + byte[] iv = new byte[12]; + random.nextBytes(iv); + return Base64.getEncoder().encodeToString(iv) + + ":" + + Base64.getEncoder() + .encodeToString(crypt(Cipher.ENCRYPT_MODE, iv, text.getBytes(StandardCharsets.UTF_8))); + } + + public String decrypt(String payload) { + String[] parts = payload.split(":", 2); + return new String( + crypt( + Cipher.DECRYPT_MODE, + Base64.getDecoder().decode(parts[0]), + Base64.getDecoder().decode(parts[1])), + StandardCharsets.UTF_8); + } + + private byte[] crypt(int mode, byte[] iv, byte[] input) { + try { + Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding"); + cipher.init(mode, new SecretKeySpec(key, "AES"), new GCMParameterSpec(128, iv)); + return cipher.doFinal(input); + } catch (GeneralSecurityException exception) { + throw new IllegalStateException("Email payload encryption failed"); + } + } +} diff --git a/src/main/java/com/beachcheck/auth/service/EmailVerificationService.java b/src/main/java/com/beachcheck/auth/service/EmailVerificationService.java index 9d968da9..9051d3e0 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailVerificationService.java +++ b/src/main/java/com/beachcheck/auth/service/EmailVerificationService.java @@ -28,7 +28,6 @@ public class EmailVerificationService { private final Logger log = LoggerFactory.getLogger(EmailVerificationService.class); - private final AsyncEmailService asyncEmailService; private final EmailVerificationTokenRepository tokenRepository; private final UserRepository userRepository; private final ApplicationEventPublisher eventPublisher; @@ -38,14 +37,12 @@ public class EmailVerificationService { private final long resendCooldownMinutes; public EmailVerificationService( - AsyncEmailService asyncEmailService, EmailVerificationTokenRepository tokenRepository, UserRepository userRepository, ApplicationEventPublisher eventPublisher, @Value("${app.email-verification.base-url}") String baseUrl, @Value("${app.email-verification.token-expiration-minutes:30}") long tokenExpirationMinutes, @Value("${app.email-verification.resend-cooldown-minutes:3}") long resendCooldownMinutes) { - this.asyncEmailService = asyncEmailService; this.tokenRepository = tokenRepository; this.userRepository = userRepository; this.eventPublisher = eventPublisher; @@ -143,8 +140,8 @@ private String hashToken(String token) { * *

Contract(Output): 쿨다운 위반 시 IllegalStateException을 던진다. * - *

Note: 이 메서드는 {@link RetryingEmailDeliveryService}의 @Retryable 백오프와는 다른 목적을 가진다. - * enforceCooldown은 사용자 행동 제어(비즈니스 규칙), @Retryable은 SMTP 장애 복구(기술 계층)이다. + *

Note: 이 메서드는 {@link EmailJobWorker}의 DB 기반 백오프와는 다른 목적을 가진다. enforceCooldown은 사용자 행동 제어(비즈니스 + * 규칙), DB 재시도는 SMTP 장애 복구(기술 계층)이다. */ private void enforceCooldown(UUID userId) { tokenRepository diff --git a/src/main/java/com/beachcheck/auth/service/RetryingEmailDeliveryService.java b/src/main/java/com/beachcheck/auth/service/RetryingEmailDeliveryService.java deleted file mode 100644 index 997b2a43..00000000 --- a/src/main/java/com/beachcheck/auth/service/RetryingEmailDeliveryService.java +++ /dev/null @@ -1,102 +0,0 @@ -package com.beachcheck.auth.service; - -import com.beachcheck.global.util.HashUtils; -import io.micrometer.tracing.Span; -import io.micrometer.tracing.Tracer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.mail.MailSendException; -import org.springframework.retry.annotation.Backoff; -import org.springframework.retry.annotation.Recover; -import org.springframework.retry.annotation.Retryable; -import org.springframework.retry.support.RetrySynchronizationManager; -import org.springframework.stereotype.Service; - -/** 비동기 이메일 작업 span 안에서 기존 SMTP 재시도 정책을 수행한다. */ -@Service -public class RetryingEmailDeliveryService { - - private static final String EMAIL_OPERATION = "verification"; - private static final String SMTP_SPAN_NAME = "email.smtp.send"; - private static final String SMTP_FAILURE_MESSAGE = "SMTP 이메일 전송 실패"; - - private static final Logger log = LoggerFactory.getLogger(RetryingEmailDeliveryService.class); - - private final EmailSender emailSender; - private final Tracer tracer; - - public RetryingEmailDeliveryService(EmailSender emailSender, Tracer tracer) { - this.emailSender = emailSender; - this.tracer = tracer; - } - - @Retryable( - retryFor = MailSendException.class, - maxAttemptsExpression = "${app.email.retry.max-attempts:4}", - backoff = - @Backoff( - delayExpression = "${app.email.retry.delay-ms:5000}", - multiplierExpression = "${app.email.retry.multiplier:2}")) - public DeliveryOutcome sendVerificationEmail( - String from, String to, String subject, String body) { - int attempt = currentAttempt(); - String recipientHash = HashUtils.sha256Hex(to); - Span smtpSpan = - tracer - .nextSpan() - .name(SMTP_SPAN_NAME) - .tag("email.operation", EMAIL_OPERATION) - .tag("email.retry.attempt", String.valueOf(attempt)) - .start(); - - try (Tracer.SpanInScope ignored = tracer.withSpan(smtpSpan)) { - log.info( - "[{}] 이메일 발송 시도 - recipientHash={}", Thread.currentThread().getName(), recipientHash); - emailSender.send(from, to, subject, body); - smtpSpan.tag("email.delivery.outcome", DeliveryOutcome.SUCCESS.value()); - log.info( - "[{}] 이메일 발송 성공 - recipientHash={}", Thread.currentThread().getName(), recipientHash); - return DeliveryOutcome.SUCCESS; - } catch (RuntimeException exception) { - smtpSpan.tag("email.delivery.outcome", "error"); - smtpSpan.tag("error.type", exception.getClass().getName()); - // 원본 SMTP 예외 메시지는 수신자 주소 등 민감정보를 포함할 수 있으므로 기록하지 않는다. - smtpSpan.error(new IllegalStateException(SMTP_FAILURE_MESSAGE)); - throw exception; - } finally { - smtpSpan.end(); - } - } - - @Recover - public DeliveryOutcome recoverFromEmailFailure( - MailSendException exception, String from, String to, String subject, String body) { - log.error( - "[{}] 이메일 발송 최종 실패 (재시도 한도 소진) - recipientHash={}", - Thread.currentThread().getName(), - HashUtils.sha256Hex(to)); - - // TODO: 향후 관리자 알림 또는 재발송 큐 추가 가능 - return DeliveryOutcome.RETRIES_EXHAUSTED; - } - - private int currentAttempt() { - var context = RetrySynchronizationManager.getContext(); - return context == null ? 1 : context.getRetryCount() + 1; - } - - public enum DeliveryOutcome { - SUCCESS("success"), - RETRIES_EXHAUSTED("retries_exhausted"); - - private final String value; - - DeliveryOutcome(String value) { - this.value = value; - } - - public String value() { - return value; - } - } -} diff --git a/src/main/java/com/beachcheck/auth/service/SmtpFailureClassifier.java b/src/main/java/com/beachcheck/auth/service/SmtpFailureClassifier.java new file mode 100644 index 00000000..55070dc6 --- /dev/null +++ b/src/main/java/com/beachcheck/auth/service/SmtpFailureClassifier.java @@ -0,0 +1,44 @@ +package com.beachcheck.auth.service; + +import jakarta.mail.MessagingException; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; +import org.eclipse.angus.mail.smtp.SMTPAddressFailedException; +import org.eclipse.angus.mail.smtp.SMTPSendFailedException; +import org.springframework.mail.MailAuthenticationException; +import org.springframework.mail.MailParseException; +import org.springframework.mail.MailPreparationException; +import org.springframework.mail.MailSendException; +import org.springframework.stereotype.Component; + +@Component +public class SmtpFailureClassifier { + public boolean retryable(Throwable failure) { + Set visited = Collections.newSetFromMap(new IdentityHashMap<>()); + return !permanent(failure, visited) && failure instanceof MailSendException; + } + + private boolean permanent(Throwable error, Set visited) { + if (error == null || !visited.add(error)) return false; + if (error instanceof jakarta.mail.AuthenticationFailedException + || error instanceof jakarta.mail.internet.AddressException) return true; + if (error instanceof MailAuthenticationException + || error instanceof MailParseException + || error instanceof MailPreparationException) return true; + if (error instanceof SMTPAddressFailedException smtp && smtp.getReturnCode() >= 500) + return true; + if (error instanceof SMTPSendFailedException smtp && smtp.getReturnCode() >= 500) return true; + if (error instanceof jakarta.mail.SendFailedException send + && send.getInvalidAddresses() != null + && send.getInvalidAddresses().length > 0) return true; + if (error instanceof MailSendException mail) { + for (Exception nested : mail.getMessageExceptions()) { + if (permanent(nested, visited)) return true; + } + } + if (error instanceof MessagingException mail && permanent(mail.getNextException(), visited)) + return true; + return permanent(error.getCause(), visited); + } +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 33eb5be5..06747b18 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -44,6 +44,9 @@ spring: properties: mail: smtp: + connectiontimeout: 5000 + timeout: 5000 + writetimeout: 5000 auth: true starttls: enable: true @@ -142,6 +145,12 @@ app: aws: region: ${AWS_REGION:us-east-1} profile: ${AWS_PROFILE:} + email: + jobs: + payload-key: ${APP_EMAIL_JOBS_PAYLOAD_KEY:} + polling-enabled: ${APP_EMAIL_JOBS_POLLING_ENABLED:true} + poll-interval-ms: ${APP_EMAIL_JOBS_POLL_INTERVAL_MS:1000} + lease-seconds: ${APP_EMAIL_JOBS_LEASE_SECONDS:300} email-verification: base-url: ${APP_EMAIL_VERIFICATION_BASE_URL:http://localhost:8080/api/auth/verify-email} token-expiration-minutes: 30 diff --git a/src/main/resources/db/migration/V15__create_email_jobs.sql b/src/main/resources/db/migration/V15__create_email_jobs.sql new file mode 100644 index 00000000..fd231abd --- /dev/null +++ b/src/main/resources/db/migration/V15__create_email_jobs.sql @@ -0,0 +1,24 @@ +CREATE TABLE email_jobs ( + id UUID PRIMARY KEY, + recipient VARCHAR(320) NOT NULL, + payload TEXT, + token_hash VARCHAR(64) NOT NULL, + producer_traceparent VARCHAR(55), + status VARCHAR(16) NOT NULL DEFAULT 'PENDING' + CHECK (status IN ('PENDING', 'SENDING', 'RETRY_WAIT', 'SUCCEEDED', 'FAILED')), + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts BETWEEN 0 AND 4), + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + last_attempt_at TIMESTAMPTZ, + next_retry_at TIMESTAMPTZ, + lease_until TIMESTAMPTZ, + last_error VARCHAR(64) +); +CREATE INDEX idx_email_jobs_due ON email_jobs (status, next_retry_at, lease_until); +CREATE TABLE email_job_attempts ( + job_id UUID NOT NULL REFERENCES email_jobs(id), + attempt INTEGER NOT NULL CHECK (attempt BETWEEN 1 AND 4), + started_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + finished_at TIMESTAMPTZ, + outcome VARCHAR(32) NOT NULL DEFAULT 'STARTED', + PRIMARY KEY (job_id, attempt) +); diff --git a/src/test/java/com/beachcheck/auth/integration/AsyncEmailRetryIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/AsyncEmailRetryIntegrationTest.java deleted file mode 100644 index 6169f206..00000000 --- a/src/test/java/com/beachcheck/auth/integration/AsyncEmailRetryIntegrationTest.java +++ /dev/null @@ -1,485 +0,0 @@ -package com.beachcheck.auth.integration; - -import static com.beachcheck.support.fixture.EmailVerificationTestFixtures.emailUser; -import static com.beachcheck.support.fixture.UniqueTestFixtures.uniqueEmail; -import static com.beachcheck.support.tracing.SpanTestSupport.flushSpans; -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.BDDMockito.then; -import static org.mockito.Mockito.after; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.doThrow; -import static org.mockito.Mockito.timeout; - -import com.beachcheck.auth.repository.EmailVerificationTokenCleanupRepository; -import com.beachcheck.auth.service.AsyncEmailService; -import com.beachcheck.auth.service.EmailSender; -import com.beachcheck.auth.service.EmailVerificationService; -import com.beachcheck.auth.service.RetryingEmailDeliveryService; -import com.beachcheck.global.util.HashUtils; -import com.beachcheck.support.base.IntegrationTest; -import com.beachcheck.support.tracing.RecordingSpanExporter; -import com.beachcheck.user.domain.User; -import com.beachcheck.user.repository.UserRepository; -import io.micrometer.tracing.Span; -import io.micrometer.tracing.Tracer; -import io.opentelemetry.api.common.AttributeKey; -import io.opentelemetry.api.trace.SpanKind; -import io.opentelemetry.api.trace.StatusCode; -import io.opentelemetry.sdk.trace.SdkTracerProvider; -import io.opentelemetry.sdk.trace.data.SpanData; -import java.net.URI; -import java.net.http.HttpClient; -import java.net.http.HttpRequest; -import java.net.http.HttpResponse; -import java.util.HashSet; -import java.util.List; -import java.util.Set; -import java.util.UUID; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.stream.IntStream; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.DisplayName; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.boot.test.context.TestConfiguration; -import org.springframework.boot.test.mock.mockito.MockBean; -import org.springframework.boot.test.mock.mockito.SpyBean; -import org.springframework.boot.test.system.CapturedOutput; -import org.springframework.boot.test.system.OutputCaptureExtension; -import org.springframework.boot.test.web.server.LocalServerPort; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Import; -import org.springframework.mail.MailAuthenticationException; -import org.springframework.mail.MailSendException; -import org.springframework.test.context.TestPropertySource; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.annotation.Propagation; -import org.springframework.transaction.annotation.Transactional; -import org.springframework.transaction.support.TransactionTemplate; - -@TestPropertySource( - properties = { - "app.email.retry.max-attempts=3", - "app.email.retry.delay-ms=10", - "app.email.retry.multiplier=1", - "management.tracing.sampling.probability=1.0" - }) -@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) -@Import(AsyncEmailRetryIntegrationTest.TraceTestConfiguration.class) -@ExtendWith(OutputCaptureExtension.class) -@DisplayName("TC4: AsyncEmailService 재시도 통합 테스트") -class AsyncEmailRetryIntegrationTest extends IntegrationTest { - - // 비동기 실행 지연을 고려해 경험적으로 잡은 여유 버퍼 - private static final int ASYNC_TIMEOUT_BUFFER_MS = 2500; - private static final int FAIL_UNTIL_ATTEMPT = 2; - - private static final String USER_EMAIL = "retry@test.com"; - private static final String VERIFICATION_LINK = - "http://localhost:8080/api/auth/verify?token=retry"; - - @Autowired private EmailVerificationService emailVerificationService; - @Autowired private EmailVerificationTokenCleanupRepository tokenCleanupRepository; - @Autowired private UserRepository userRepository; - @Autowired private PlatformTransactionManager transactionManager; - @Autowired private Tracer tracer; - @Autowired private SdkTracerProvider tracerProvider; - @Autowired private RecordingSpanExporter spanExporter; - @LocalServerPort private int serverPort; - - @Value("${app.email.retry.max-attempts}") - private int retryMaxAttempts; - - @Value("${app.email.retry.delay-ms}") - private long retryDelayMs; - - @Autowired private AsyncEmailService asyncEmailService; - @SpyBean private RetryingEmailDeliveryService retryingEmailDeliveryService; - @MockBean private EmailSender emailSender; - - private final Set createdUserIds = new HashSet<>(); - private TransactionTemplate transactionTemplate; - private int asyncTimeoutMs; - - @BeforeEach - void setUp() { - transactionTemplate = new TransactionTemplate(transactionManager); - asyncTimeoutMs = calculateAsyncTimeoutMs(); - flushSpans(tracerProvider); - spanExporter.clear(); - } - - @Test - @DisplayName("TC4-01: 메일 전송이 일시 실패하면 재시도 후 성공한다") - void sendVerificationEmailAsync_retryThenSuccess(CapturedOutput output) { - // given - givenEmailSenderFailsThenSucceeds(); - - // when - asyncEmailService.sendVerificationEmailAsync(USER_EMAIL, VERIFICATION_LINK); - - // then - assertRetriedSendCountTo(USER_EMAIL, retryMaxAttempts); - assertRecoverNotCalled(); - assertDeliverySpans( - awaitEmailSpans(FAIL_UNTIL_ATTEMPT + 1), - retryMaxAttempts, - "success", - "error", - "error", - "success"); - assertRecipientLogSafe(output, USER_EMAIL); - } - - @Test - @DisplayName("TC4-02: 최대 재시도 소진 시 recover 경로로 종료된다") - void sendVerificationEmailAsync_retryExhausted_thenRecover(CapturedOutput output) { - // given - givenEmailSenderAlwaysFails(); - - // when - asyncEmailService.sendVerificationEmailAsync(USER_EMAIL, VERIFICATION_LINK); - - // then - assertRetriedSendCountTo(USER_EMAIL, retryMaxAttempts); - assertRecoverCalledOnceFor(USER_EMAIL, VERIFICATION_LINK); - assertThat(output.getOut()) - .contains("이메일 발송 최종 실패 (재시도 한도 소진) - recipientHash=" + HashUtils.sha256Hex(USER_EMAIL)); - assertRecipientLogSafe(output, USER_EMAIL); - assertDeliverySpans( - awaitEmailSpans(retryMaxAttempts), - retryMaxAttempts, - "retries_exhausted", - "error", - "error", - "error"); - } - - @Test - @DisplayName("TC4-03: 비재시도 메일 예외도 원문 민감정보 없이 안전하게 기록한다") - void sendVerificationEmailAsync_nonRetryableFailure_doesNotLeakSensitiveLog( - CapturedOutput output) { - String sensitiveExceptionMessage = - "recipient=" + USER_EMAIL + ", verificationLink=" + VERIFICATION_LINK; - doThrow(new MailAuthenticationException(sensitiveExceptionMessage)) - .when(emailSender) - .send(anyString(), anyString(), anyString(), anyString()); - - asyncEmailService.sendVerificationEmailAsync(USER_EMAIL, VERIFICATION_LINK); - - assertRetriedSendCountTo(USER_EMAIL, 1); - assertRecoverNotCalled(); - assertThat(output.getOut()) - .contains("비동기 작업 처리 실패") - .contains("errorType=" + MailAuthenticationException.class.getName()) - .doesNotContain(sensitiveExceptionMessage); - assertRecipientLogSafe(output, USER_EMAIL); - assertNonRetryableFailureSpan(awaitEmailSpans(1)); - } - - @Test - @Transactional(propagation = Propagation.NOT_SUPPORTED) - @DisplayName("TC4-04: AFTER_COMMIT 이벤트 리스너 경로에서도 재시도 후 성공한다") - void sendVerification_viaEventListener_retryThenSuccess() { - // given - User user = saveUser(); - givenEmailSenderFailsThenSucceeds(); - - // when - transactionTemplate.executeWithoutResult( - unused -> { - User managedUser = userRepository.findById(user.getId()).orElseThrow(); - emailVerificationService.sendVerification(managedUser); - }); - - // then - assertRetriedSendCountTo(user.getEmail(), retryMaxAttempts); - assertRecoverNotCalled(); - } - - @Test - @Transactional(propagation = Propagation.NOT_SUPPORTED) - @DisplayName("TC4-05: AFTER_COMMIT 부모 Trace가 비동기 이메일 작업 span으로 이어진다") - void sendVerification_afterCommit_continuesTraceIntoAsyncEmail() { - User user = saveUser(); - Span parent = tracer.nextSpan().name("email.after-commit.parent").start(); - - try (Tracer.SpanInScope ignored = tracer.withSpan(parent)) { - transactionTemplate.executeWithoutResult( - unused -> { - User managedUser = userRepository.findById(user.getId()).orElseThrow(); - emailVerificationService.sendVerification(managedUser); - }); - } finally { - parent.end(); - } - - assertRetriedSendCountTo(user.getEmail(), 1); - List spans = awaitEmailSpans(1); - SpanData parentSpan = onlySpanNamed(spans, "email.after-commit.parent"); - SpanData workSpan = onlySpanNamed(spans, "email.verification.send"); - - assertThat(workSpan.getTraceId()).isEqualTo(parentSpan.getTraceId()); - assertThat(workSpan.getParentSpanId()).isEqualTo(parentSpan.getSpanId()); - assertDeliverySpans(spans, 1, "success", "success"); - } - - @Test - @Transactional(propagation = Propagation.NOT_SUPPORTED) - @DisplayName("TC4-06: 실제 회원가입 HTTP Trace가 AFTER_COMMIT 비동기 이메일 span으로 이어진다") - void signUpHttpRequest_afterCommit_continuesTraceIntoAsyncEmail(CapturedOutput output) - throws Exception { - String email = uniqueEmail("trace-http-signup"); - String requestBody = - """ - {"email":"%s","password":"Password1!","name":"테스트 사용자"} - """ - .formatted(email); - HttpRequest request = - HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + serverPort + "/api/auth/signup")) - .header("Content-Type", "application/json") - .POST(HttpRequest.BodyPublishers.ofString(requestBody)) - .build(); - - HttpResponse response = - HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString()); - - User savedUser = userRepository.findByEmail(email).orElseThrow(); - createdUserIds.add(savedUser.getId()); - assertThat(response.statusCode()).isEqualTo(201); - assertRetriedSendCountTo(email, 1); - - List spans = awaitEmailSpans(1); - SpanData workSpan = onlySpanNamed(spans, "email.verification.send"); - SpanData serverSpan = httpServerAncestorOf(spans, workSpan); - - assertThat(workSpan.getTraceId()).isEqualTo(serverSpan.getTraceId()); - assertDeliverySpans(spans, 1, "success", "success"); - assertRecipientLogSafe(output, email); - } - - private void assertRecipientLogSafe(CapturedOutput output, String recipientEmail) { - assertThat(output.getOut()) - .contains("recipientHash=" + HashUtils.sha256Hex(recipientEmail)) - .doesNotContain(recipientEmail) - .doesNotContain(VERIFICATION_LINK) - .doesNotContain("?token=") - .doesNotContain("이메일 인증"); - } - - private void assertDeliverySpans( - List spans, int expectedSmtpSpanCount, String workOutcome, String... smtpOutcomes) { - SpanData workSpan = onlySpanNamed(spans, "email.verification.send"); - List smtpSpans = - spans.stream() - .filter(span -> span.getName().equals("email.smtp.send")) - .sorted( - (left, right) -> - Integer.compare( - Integer.parseInt(attribute(left, "email.retry.attempt")), - Integer.parseInt(attribute(right, "email.retry.attempt")))) - .toList(); - - assertThat(smtpSpans).hasSize(expectedSmtpSpanCount); - assertThat(attribute(workSpan, "email.operation")).isEqualTo("verification"); - assertThat(attribute(workSpan, "email.delivery.outcome")).isEqualTo(workOutcome); - List expectedAttempts = - IntStream.rangeClosed(1, expectedSmtpSpanCount).mapToObj(String::valueOf).toList(); - assertThat(smtpSpans) - .extracting(span -> attribute(span, "email.retry.attempt")) - .containsExactlyElementsOf(expectedAttempts); - assertThat(smtpSpans) - .extracting(span -> attribute(span, "email.delivery.outcome")) - .containsExactly(smtpOutcomes); - assertThat(smtpSpans) - .allSatisfy( - span -> { - assertThat(span.getTraceId()).isEqualTo(workSpan.getTraceId()); - assertThat(span.getParentSpanId()).isEqualTo(workSpan.getSpanId()); - assertThat(attribute(span, "email.operation")).isEqualTo("verification"); - }); - - List errorSpans = - smtpSpans.stream() - .filter(span -> attribute(span, "email.delivery.outcome").equals("error")) - .toList(); - assertThat(errorSpans) - .allSatisfy( - span -> assertThat(span.getStatus().getStatusCode()).isEqualTo(StatusCode.ERROR)); - assertNoSensitiveData(spans); - } - - private void assertNonRetryableFailureSpan(List spans) { - SpanData workSpan = onlySpanNamed(spans, "email.verification.send"); - SpanData smtpSpan = onlySpanNamed(spans, "email.smtp.send"); - - assertThat(smtpSpan.getTraceId()).isEqualTo(workSpan.getTraceId()); - assertThat(smtpSpan.getParentSpanId()).isEqualTo(workSpan.getSpanId()); - assertThat(attribute(smtpSpan, "email.retry.attempt")).isEqualTo("1"); - assertThat(attribute(smtpSpan, "email.delivery.outcome")).isEqualTo("error"); - assertThat(smtpSpan.getStatus().getStatusCode()).isEqualTo(StatusCode.ERROR); - assertNoSensitiveData(spans); - } - - private void assertNoSensitiveData(List spans) { - String[] sensitiveValues = {USER_EMAIL, VERIFICATION_LINK, "이메일 인증", "request-id", "user-id"}; - assertThat(spans) - .allSatisfy( - span -> { - String attributes = span.getAttributes().toString(); - assertThat(span.getName()).doesNotContain(sensitiveValues); - assertThat(attributes).doesNotContain(sensitiveValues); - assertThat(span.getEvents()) - .allSatisfy( - event -> { - assertThat(event.getName()).doesNotContain(sensitiveValues); - assertThat(event.getAttributes().toString()) - .doesNotContain(sensitiveValues); - }); - }); - } - - private List awaitEmailSpans(int expectedSmtpSpanCount) { - for (int attempt = 0; attempt < 100; attempt++) { - flushSpans(tracerProvider); - List spans = spanExporter.spans(); - long smtpSpanCount = - spans.stream().filter(span -> span.getName().equals("email.smtp.send")).count(); - boolean workSpanPresent = - spans.stream().anyMatch(span -> span.getName().equals("email.verification.send")); - if (workSpanPresent && smtpSpanCount >= expectedSmtpSpanCount) { - return spans; - } - try { - Thread.sleep(10); - } catch (InterruptedException exception) { - Thread.currentThread().interrupt(); - throw new IllegalStateException("이메일 Trace 대기가 중단되었습니다.", exception); - } - } - return spanExporter.spans(); - } - - private SpanData httpServerAncestorOf(List spans, SpanData descendant) { - String parentSpanId = descendant.getParentSpanId(); - for (int depth = 0; depth < spans.size(); depth++) { - String expectedSpanId = parentSpanId; - SpanData parent = - spans.stream() - .filter(span -> span.getSpanId().equals(expectedSpanId)) - .findFirst() - .orElseThrow(); - if (parent.getKind() == SpanKind.SERVER) { - return parent; - } - parentSpanId = parent.getParentSpanId(); - } - throw new AssertionError("이메일 작업 span의 상위에서 HTTP server span을 찾지 못했습니다."); - } - - private void assertRetriedSendCountTo(String recipientEmail, int expectedCount) { - then(emailSender) - .should(timeout(asyncTimeoutMs).times(expectedCount)) - .send(anyString(), eq(recipientEmail), anyString(), anyString()); - then(emailSender) - .should(after(asyncTimeoutMs).times(expectedCount)) - .send(anyString(), eq(recipientEmail), anyString(), anyString()); - } - - private void assertRecoverCalledOnceFor(String recipientEmail, String verificationLink) { - then(retryingEmailDeliveryService) - .should(timeout(asyncTimeoutMs).times(1)) - .recoverFromEmailFailure( - any(MailSendException.class), - anyString(), - eq(recipientEmail), - anyString(), - anyString()); - then(retryingEmailDeliveryService) - .should(after(asyncTimeoutMs).times(1)) - .recoverFromEmailFailure( - any(MailSendException.class), - anyString(), - eq(recipientEmail), - anyString(), - anyString()); - } - - private void assertRecoverNotCalled() { - then(retryingEmailDeliveryService) - .should(after(asyncTimeoutMs).never()) - .recoverFromEmailFailure( - any(MailSendException.class), anyString(), anyString(), anyString(), anyString()); - } - - private void givenEmailSenderFailsThenSucceeds() { - AtomicInteger attempts = new AtomicInteger(0); - doAnswer( - invocation -> { - if (attempts.incrementAndGet() <= FAIL_UNTIL_ATTEMPT) { - throw new MailSendException("일시적 SMTP 전송 실패"); - } - return null; - }) - .when(emailSender) - .send(anyString(), anyString(), anyString(), anyString()); - } - - private void givenEmailSenderAlwaysFails() { - doThrow(new MailSendException("지속적 SMTP 전송 실패")) - .when(emailSender) - .send(anyString(), anyString(), anyString(), anyString()); - } - - private User saveUser() { - User user = emailUser(uniqueEmail("retry-listener"), false); - User saved = userRepository.save(user); - createdUserIds.add(saved.getId()); - return saved; - } - - @AfterEach - void cleanUp() { - if (createdUserIds.isEmpty()) { - return; - } - - transactionTemplate.executeWithoutResult( - unused -> { - entityManager.clear(); - tokenCleanupRepository.deleteAllByUserIds(createdUserIds); - userRepository.deleteAllByIdInBatch(createdUserIds); - }); - createdUserIds.clear(); - } - - private int calculateAsyncTimeoutMs() { - return (retryMaxAttempts - 1) * (int) retryDelayMs + ASYNC_TIMEOUT_BUFFER_MS; - } - - private SpanData onlySpanNamed(List spans, String name) { - List matching = spans.stream().filter(span -> span.getName().equals(name)).toList(); - assertThat(matching).hasSize(1); - return matching.getFirst(); - } - - private String attribute(SpanData span, String key) { - return span.getAttributes().get(AttributeKey.stringKey(key)); - } - - @TestConfiguration(proxyBeanMethods = false) - static class TraceTestConfiguration { - - @Bean - RecordingSpanExporter recordingSpanExporter() { - return new RecordingSpanExporter(); - } - } -} diff --git a/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java new file mode 100644 index 00000000..caebb724 --- /dev/null +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java @@ -0,0 +1,350 @@ +package com.beachcheck.auth.integration; + +import static com.beachcheck.support.fixture.EmailVerificationTestFixtures.emailUser; +import static com.beachcheck.support.fixture.UniqueTestFixtures.uniqueEmail; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; + +import com.beachcheck.auth.repository.EmailJobStore; +import com.beachcheck.auth.repository.EmailJobStore.Job; +import com.beachcheck.auth.service.EmailJobScheduler; +import com.beachcheck.auth.service.EmailJobWorker; +import com.beachcheck.auth.service.EmailPayloadCipher; +import com.beachcheck.auth.service.EmailSender; +import com.beachcheck.auth.service.EmailVerificationService; +import com.beachcheck.auth.service.SmtpFailureClassifier; +import com.beachcheck.support.base.IntegrationTest; +import com.beachcheck.user.domain.User; +import com.beachcheck.user.repository.UserRepository; +import io.micrometer.tracing.Tracer; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.mail.MailAuthenticationException; +import org.springframework.mail.MailSendException; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; + +@Transactional(propagation = Propagation.NOT_SUPPORTED) +class EmailJobIntegrationTest extends IntegrationTest { + @Autowired private EmailVerificationService verification; + @Autowired private UserRepository users; + @Autowired private JdbcTemplate jdbc; + @Autowired private EmailJobWorker worker; + @Autowired private EmailPayloadCipher cipher; + @Autowired private EmailJobStore store; + @MockBean private EmailSender sender; + private User user; + private UUID jobId; + + @BeforeEach + void prepare() { + jdbc.update("DELETE FROM email_job_attempts"); + jdbc.update("DELETE FROM email_jobs"); + user = users.save(emailUser(uniqueEmail("durable"), false)); + verification.sendVerification(user); + jobId = + jdbc.queryForObject( + "SELECT id FROM email_jobs WHERE recipient = ?", UUID.class, user.getEmail()); + } + + @AfterEach + void cleanup() { + jdbc.update("DELETE FROM email_job_attempts"); + jdbc.update("DELETE FROM email_jobs"); + users.deleteById(user.getId()); + } + + @Test + void pendingSurvivesWorkerRecreationAndSuccessIsNotSentAgain() { + var restarted = + new EmailJobWorker(store, sender, cipher, new SmtpFailureClassifier(), Tracer.NOOP, ""); + new EmailJobScheduler(restarted).poll(); + worker.poll(); + assertThat(status()).isEqualTo("SUCCEEDED"); + assertThat(attempts()).isEqualTo(1); + verify(sender, times(1)).send(anyString(), eq(user.getEmail()), anyString(), anyString()); + assertThat( + jdbc.queryForObject("SELECT payload FROM email_jobs WHERE id = ?", String.class, jobId)) + .isNull(); + assertThat(history()).containsExactly("ACCEPTED"); + } + + @Test + void staleSendingRecoversAndOldWorkerCannotOverwriteNewClaim() { + Job old = store.claim().orElseThrow(); + assertThat(store.claim()).isEmpty(); + expireLease(); + Job recovered = store.claim().orElseThrow(); + assertThat(recovered.attempt()).isEqualTo(2); + assertThat(store.finish(old, "SUCCEEDED", "ACCEPTED", 0)).isFalse(); + assertThat(status()).isEqualTo("SENDING"); + assertThat(store.finish(recovered, "SUCCEEDED", "ACCEPTED", 0)).isTrue(); + assertThat(history()).containsExactly("UNKNOWN", "ACCEPTED"); + } + + @Test + void concurrentWorkersOnlySendOnce() throws Exception { + CountDownLatch start = new CountDownLatch(1); + try (var pool = Executors.newFixedThreadPool(2)) { + var first = + pool.submit( + () -> { + start.await(); + worker.poll(); + return true; + }); + var secondWorker = + new EmailJobWorker(store, sender, cipher, new SmtpFailureClassifier(), Tracer.NOOP, ""); + var second = + pool.submit( + () -> { + start.await(); + secondWorker.poll(); + return true; + }); + start.countDown(); + first.get(10, TimeUnit.SECONDS); + second.get(10, TimeUnit.SECONDS); + } + verify(sender, times(1)).send(anyString(), eq(user.getEmail()), anyString(), anyString()); + assertThat(attempts()).isEqualTo(1); + } + + @Test + void smtpAcceptanceThenDatabaseFailureLeavesUncertainAttemptAndMayDuplicate() { + jdbc.execute( + "ALTER TABLE email_job_attempts ADD CONSTRAINT test_reject_accepted CHECK (outcome <> 'ACCEPTED') NOT VALID"); + try { + assertThatThrownBy(worker::poll) + .isInstanceOf(org.springframework.dao.DataAccessException.class); + } finally { + jdbc.execute("ALTER TABLE email_job_attempts DROP CONSTRAINT test_reject_accepted"); + } + assertThat(status()).isEqualTo("SENDING"); + assertThat(history()).containsExactly("STARTED"); + worker.poll(); + verify(sender, times(1)).send(anyString(), anyString(), anyString(), anyString()); + expireLease(); + worker.poll(); + assertThat(status()).isEqualTo("SUCCEEDED"); + verify(sender, times(2)).send(anyString(), anyString(), anyString(), anyString()); + assertThat(history()).containsExactly("UNKNOWN", "ACCEPTED"); + } + + @Test + void retryBudgetAndExponentialBackoffAreDurableAndDoNotResetCooldown() { + doThrow(new MailSendException("transient")) + .when(sender) + .send(anyString(), anyString(), anyString(), anyString()); + var tokenCreatedAt = + jdbc.queryForObject( + "SELECT created_at FROM email_verification_tokens WHERE user_id = ?", + java.sql.Timestamp.class, + user.getId()); + for (int attempt = 1; attempt <= 4; attempt++) { + worker.poll(); + assertThat(attempts()).isEqualTo(attempt); + if (attempt < 4) { + assertThat(status()).isEqualTo("RETRY_WAIT"); + double delay = + jdbc.queryForObject( + "SELECT EXTRACT(EPOCH FROM (next_retry_at - last_attempt_at)) FROM email_jobs WHERE id = ?", + Double.class, + jobId); + assertThat(delay) + .isBetween((double) (5 << (attempt - 1)), (double) (5 << (attempt - 1)) + 3); + worker.poll(); // 아직 기한 전: SMTP 및 횟수 증가 없음 + assertThat(attempts()).isEqualTo(attempt); + jdbc.update( + "UPDATE email_jobs SET next_retry_at = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE id = ?", + jobId); + } + } + assertThat(status()).isEqualTo("FAILED"); + worker.poll(); + verify(sender, times(4)).send(anyString(), anyString(), anyString(), anyString()); + assertThat(history()).hasSize(4); + assertThat( + jdbc.queryForObject( + "SELECT created_at FROM email_verification_tokens WHERE user_id = ?", + java.sql.Timestamp.class, + user.getId())) + .isEqualTo(tokenCreatedAt); + assertThatThrownBy(() -> verification.resendVerification(user.getEmail())) + .isInstanceOf(IllegalStateException.class); + jdbc.update( + "UPDATE email_verification_tokens SET created_at = CURRENT_TIMESTAMP - INTERVAL '4 minutes' WHERE user_id = ?", + user.getId()); + verification.resendVerification(user.getEmail()); + assertThat( + jdbc.queryForObject( + "SELECT count(*) FROM email_jobs WHERE recipient = ?", + Integer.class, + user.getEmail())) + .isEqualTo(2); + assertThat(status()).isEqualTo("FAILED"); + } + + @Test + void permanentFailureDoesNotRetry() { + doThrow(new MailAuthenticationException("bad credentials")) + .when(sender) + .send(anyString(), anyString(), anyString(), anyString()); + worker.poll(); + worker.poll(); + assertThat(status()).isEqualTo("FAILED"); + assertThat(attempts()).isEqualTo(1); + assertThat(history()).containsExactly("PERMANENT_FAILURE"); + } + + @Test + void fourthAbandonedClaimNeverGetsFifthAttempt() { + for (int attempt = 1; attempt <= 4; attempt++) { + assertThat(store.claim().orElseThrow().attempt()).isEqualTo(attempt); + expireLease(); + } + worker.poll(); + assertThat(status()).isEqualTo("FAILED"); + assertThat(attempts()).isEqualTo(4); + assertThat(history()).containsExactly("UNKNOWN", "UNKNOWN", "UNKNOWN", "UNKNOWN"); + verifyNoInteractions(sender); + } + + @Test + void expiredAndSupersededTokensAreNotDelivered() { + jdbc.update( + "UPDATE email_verification_tokens SET expires_at = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE user_id = ?", + user.getId()); + worker.poll(); + assertThat(status()).isEqualTo("FAILED"); + assertThat(attempts()).isZero(); + verifyNoInteractions(sender); + } + + @Test + void closingAndStartingApplicationContextsRecoversPendingAndStaleSending() { + try (var first = restartContext()) { + first.getBean(EmailJobStore.class).claim().orElseThrow(); + } + expireLease(); + verification.sendVerification(user); // 다른 PENDING 작업도 재시작 후 처리 + try (var restarted = restartContext()) { + restarted.publishEvent( + new org.springframework.boot.context.event.ApplicationReadyEvent( + new org.springframework.boot.SpringApplication(), + new String[0], + restarted, + java.time.Duration.ZERO)); + } + assertThat( + jdbc.queryForObject( + "SELECT count(*) FROM email_jobs WHERE status = 'SUCCEEDED'", Integer.class)) + .isEqualTo(2); + assertThat(attempts()).isEqualTo(2); + verify(sender, times(2)).send(anyString(), anyString(), anyString(), anyString()); + } + + private org.springframework.context.annotation.AnnotationConfigApplicationContext + restartContext() { + var context = new org.springframework.context.annotation.AnnotationConfigApplicationContext(); + context.register(RestartTransactions.class); + context.registerBean( + org.springframework.transaction.PlatformTransactionManager.class, + () -> + new org.springframework.jdbc.datasource.DataSourceTransactionManager( + jdbc.getDataSource())); + context.registerBean( + EmailJobStore.class, () -> new EmailJobStore(new JdbcTemplate(jdbc.getDataSource()), 300)); + context.registerBean( + EmailJobWorker.class, + () -> + new EmailJobWorker( + context.getBean(EmailJobStore.class), + sender, + cipher, + new SmtpFailureClassifier(), + Tracer.NOOP, + "")); + context.registerBean( + EmailJobScheduler.class, + () -> new EmailJobScheduler(context.getBean(EmailJobWorker.class))); + context.refresh(); + return context; + } + + @org.springframework.boot.test.context.TestConfiguration(proxyBeanMethods = false) + @org.springframework.transaction.annotation.EnableTransactionManagement + static class RestartTransactions {} + + @Test + void enqueueFailureRollsBackNewTokenAsWell() { + int before = + jdbc.queryForObject( + "SELECT count(*) FROM email_verification_tokens WHERE user_id = ?", + Integer.class, + user.getId()); + jdbc.execute( + "ALTER TABLE email_jobs ADD CONSTRAINT test_reject_pending CHECK (status <> 'PENDING') NOT VALID"); + try { + assertThatThrownBy(() -> verification.sendVerification(user)) + .isInstanceOf(org.springframework.dao.DataAccessException.class); + } finally { + jdbc.execute("ALTER TABLE email_jobs DROP CONSTRAINT test_reject_pending"); + } + assertThat( + jdbc.queryForObject( + "SELECT count(*) FROM email_verification_tokens WHERE user_id = ?", + Integer.class, + user.getId())) + .isEqualTo(before); + } + + @Test + void manualResendInvalidatesOldPendingJobWithoutResettingItsAttempts() { + jdbc.update( + "UPDATE email_verification_tokens SET created_at = CURRENT_TIMESTAMP - INTERVAL '4 minutes' WHERE user_id = ?", + user.getId()); + verification.resendVerification(user.getEmail()); + worker.poll(); + worker.poll(); + assertThat(status()).isEqualTo("FAILED"); + assertThat(attempts()).isZero(); + verify(sender, times(1)).send(anyString(), eq(user.getEmail()), anyString(), anyString()); + } + + private void expireLease() { + jdbc.update( + "UPDATE email_jobs SET lease_until = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE id = ?", + jobId); + } + + private String status() { + return jdbc.queryForObject("SELECT status FROM email_jobs WHERE id = ?", String.class, jobId); + } + + private int attempts() { + return jdbc.queryForObject( + "SELECT attempts FROM email_jobs WHERE id = ?", Integer.class, jobId); + } + + private java.util.List history() { + return jdbc.queryForList( + "SELECT outcome FROM email_job_attempts WHERE job_id = ? ORDER BY attempt", + String.class, + jobId); + } +} diff --git a/src/test/java/com/beachcheck/auth/integration/AsyncEmailTracingIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java similarity index 71% rename from src/test/java/com/beachcheck/auth/integration/AsyncEmailTracingIntegrationTest.java rename to src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java index 00454f95..cbb3a30d 100644 --- a/src/test/java/com/beachcheck/auth/integration/AsyncEmailTracingIntegrationTest.java +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java @@ -4,9 +4,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import com.beachcheck.auth.service.AsyncEmailService; import com.beachcheck.auth.service.EmailSender; -import com.beachcheck.auth.service.RetryingEmailDeliveryService; import com.beachcheck.global.config.AsyncConfig; import com.beachcheck.global.config.RetryConfig; import com.beachcheck.global.logging.MdcTaskDecorator; @@ -37,7 +35,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; @DisplayName("비동기 이메일 Trace 통합 계약") -class AsyncEmailTracingIntegrationTest { +class EmailJobTracingIntegrationTest { private static final String USER_EMAIL = "user@test.com"; private static final String VERIFICATION_LINK = "https://example.com/verify?token=test-token"; @@ -58,65 +56,62 @@ class AsyncEmailTracingIntegrationTest { .withPropertyValues("management.tracing.sampling.probability=1.0"); @Test - @DisplayName("부모 Trace를 worker로 전달하고 이메일 작업 자식 span을 하나 만든다") - void parentTrace_createsChildWorkSpan() { + @DisplayName("DB에서 복구한 이메일 시도는 생산 Trace에 Link를 연결한다") + void recoveredJobLinksProducerAndDoesNotLeakSensitiveData() { contextRunner.run( context -> { Tracer tracer = context.getBean(Tracer.class); - AsyncEmailService asyncEmailService = context.getBean(AsyncEmailService.class); - RecordingEmailSender emailSender = context.getBean(RecordingEmailSender.class); - RecordingSpanExporter exporter = context.getBean(RecordingSpanExporter.class); - SdkTracerProvider tracerProvider = context.getBean(SdkTracerProvider.class); - emailSender.prepare(); - + var store = org.mockito.Mockito.mock(com.beachcheck.auth.repository.EmailJobStore.class); + var cipher = + new com.beachcheck.auth.service.EmailPayloadCipher( + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="); Span parent = tracer.nextSpan().name("http.parent.test").start(); - try (Tracer.SpanInScope ignored = tracer.withSpan(parent)) { - asyncEmailService.sendVerificationEmailAsync(USER_EMAIL, VERIFICATION_LINK); - } finally { - parent.end(); - } - - assertThat(emailSender.await()).isTrue(); - List spans = awaitSpans(exporter, tracerProvider, 3); - SpanData parentData = onlySpanNamed(spans, "http.parent.test"); - SpanData workSpan = onlySpanNamed(spans, "email.verification.send"); - SpanData smtpSpan = onlySpanNamed(spans, "email.smtp.send"); - - assertThat(workSpan.getTraceId()).isEqualTo(parentData.getTraceId()); - assertThat(workSpan.getParentSpanId()).isEqualTo(parentData.getSpanId()); - assertThat(attribute(workSpan, "email.operation")).isEqualTo("verification"); - assertThat(attribute(workSpan, "email.delivery.outcome")).isEqualTo("success"); - assertThat(smtpSpan.getTraceId()).isEqualTo(workSpan.getTraceId()); - assertThat(smtpSpan.getParentSpanId()).isEqualTo(workSpan.getSpanId()); - assertThat(attribute(smtpSpan, "email.operation")).isEqualTo("verification"); - assertThat(attribute(smtpSpan, "email.retry.attempt")).isEqualTo("1"); - assertThat(attribute(smtpSpan, "email.delivery.outcome")).isEqualTo("success"); - }); - } - - @Test - @DisplayName("부모 Trace가 없으면 이메일 작업을 새 루트 span으로 시작한다") - void noParent_createsRootWorkSpan() { - contextRunner.run( - context -> { - AsyncEmailService asyncEmailService = context.getBean(AsyncEmailService.class); - RecordingEmailSender emailSender = context.getBean(RecordingEmailSender.class); - RecordingSpanExporter exporter = context.getBean(RecordingSpanExporter.class); - SdkTracerProvider tracerProvider = context.getBean(SdkTracerProvider.class); - emailSender.prepare(); - - asyncEmailService.sendVerificationEmailAsync(USER_EMAIL, VERIFICATION_LINK); - - assertThat(emailSender.await()).isTrue(); - List spans = awaitSpans(exporter, tracerProvider, 2); - SpanData workSpan = onlySpanNamed(spans, "email.verification.send"); - SpanData smtpSpan = onlySpanNamed(spans, "email.smtp.send"); - - assertThat(workSpan.getParentSpanContext().isValid()).isFalse(); - assertThat(attribute(workSpan, "email.delivery.outcome")).isEqualTo("success"); - assertThat(smtpSpan.getParentSpanId()).isEqualTo(workSpan.getSpanId()); - assertThat(attribute(smtpSpan, "email.retry.attempt")).isEqualTo("1"); - assertThat(attribute(smtpSpan, "email.delivery.outcome")).isEqualTo("success"); + String traceparent = + "00-" + parent.context().traceId() + "-" + parent.context().spanId() + "-01"; + parent.end(); + var job = + new com.beachcheck.auth.repository.EmailJobStore.Job( + java.util.UUID.randomUUID(), + USER_EMAIL, + cipher.encrypt(VERIFICATION_LINK), + 3, + traceparent); + org.mockito.Mockito.when(store.claim()) + .thenReturn(java.util.Optional.of(job), java.util.Optional.empty()); + org.mockito.Mockito.when( + store.finish( + org.mockito.ArgumentMatchers.any(), + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyLong())) + .thenReturn(true); + var worker = + new com.beachcheck.auth.service.EmailJobWorker( + store, + (from, to, subject, body) -> { + throw new org.springframework.mail.MailSendException(VERIFICATION_LINK); + }, + cipher, + new com.beachcheck.auth.service.SmtpFailureClassifier(), + tracer, + "sender@example.com"); + worker.poll(); + var spans = + awaitSpans( + context.getBean(RecordingSpanExporter.class), + context.getBean(SdkTracerProvider.class), + 3); + var work = onlySpanNamed(spans, "email.verification.send"); + var smtp = onlySpanNamed(spans, "email.smtp.send"); + assertThat(work.getParentSpanContext().isValid()).isFalse(); + assertThat(work.getLinks()).hasSize(1); + assertThat(work.getLinks().getFirst().getSpanContext().getTraceId()) + .isEqualTo(parent.context().traceId()); + assertThat(smtp.getParentSpanId()).isEqualTo(work.getSpanId()); + assertThat(attribute(smtp, "email.retry.attempt")).isEqualTo("3"); + assertThat(attribute(work, "email.delivery.outcome")).isEqualTo("retry_wait"); + assertThat(spans.toString()).doesNotContain(USER_EMAIL, VERIFICATION_LINK, "?token="); + assertThat(tracer.currentSpan()).isNull(); }); } @@ -262,18 +257,6 @@ RecordingSpanExporter recordingSpanExporter() { RecordingEmailSender emailSender() { return new RecordingEmailSender(); } - - @Bean - RetryingEmailDeliveryService retryingEmailDeliveryService( - EmailSender emailSender, Tracer tracer) { - return new RetryingEmailDeliveryService(emailSender, tracer); - } - - @Bean - AsyncEmailService asyncEmailService( - RetryingEmailDeliveryService emailDeliveryService, Tracer tracer) { - return new AsyncEmailService(emailDeliveryService, tracer, "no-reply@test.com", 30); - } } static class RecordingEmailSender implements EmailSender { diff --git a/src/test/java/com/beachcheck/auth/integration/EmailAfterCommitIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobTransactionIntegrationTest.java similarity index 70% rename from src/test/java/com/beachcheck/auth/integration/EmailAfterCommitIntegrationTest.java rename to src/test/java/com/beachcheck/auth/integration/EmailJobTransactionIntegrationTest.java index a78eb04b..ba49c414 100644 --- a/src/test/java/com/beachcheck/auth/integration/EmailAfterCommitIntegrationTest.java +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobTransactionIntegrationTest.java @@ -3,10 +3,8 @@ import static com.beachcheck.support.fixture.EmailVerificationTestFixtures.emailUser; import static com.beachcheck.support.fixture.UniqueTestFixtures.uniqueEmail; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.BDDMockito.then; import com.beachcheck.auth.repository.EmailVerificationTokenRepository; -import com.beachcheck.auth.service.AsyncEmailService; import com.beachcheck.auth.service.EmailVerificationService; import com.beachcheck.support.base.IntegrationTest; import com.beachcheck.user.domain.User; @@ -18,23 +16,22 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; -import org.mockito.ArgumentCaptor; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionTemplate; @Transactional(propagation = Propagation.NOT_SUPPORTED) -class EmailAfterCommitIntegrationTest extends IntegrationTest { +class EmailJobTransactionIntegrationTest extends IntegrationTest { @Autowired private EmailVerificationService emailVerificationService; @Autowired private UserRepository userRepository; @Autowired private EmailVerificationTokenRepository tokenRepository; @Autowired private PlatformTransactionManager transactionManager; - @MockBean private AsyncEmailService asyncEmailService; + @Autowired private JdbcTemplate jdbc; private final List createdUserIds = new ArrayList<>(); private TransactionTemplate transactionTemplate; @@ -49,6 +46,11 @@ void cleanUp() { transactionTemplate.executeWithoutResult( unused -> { if (!createdUserIds.isEmpty()) { + for (UUID id : createdUserIds) { + jdbc.update( + "DELETE FROM email_jobs WHERE recipient IN (SELECT email FROM users WHERE id = ?)", + id); + } entityManager .createQuery("delete from EmailVerificationToken t where t.user.id in :userIds") .setParameter("userIds", createdUserIds) @@ -60,8 +62,8 @@ void cleanUp() { } @Test - @DisplayName("커밋 후 이메일 비동기 전송이 호출된다") - void sendVerification_afterCommit_triggersListener() { + @DisplayName("커밋 시 인증 토큰과 이메일 작업이 함께 저장된다") + void sendVerification_commit_persistsTokenAndJob() { // given User user = saveUser(); @@ -71,17 +73,17 @@ void sendVerification_afterCommit_triggersListener() { User managedUser = userRepository.findById(user.getId()).orElseThrow(); emailVerificationService.sendVerification(managedUser); - // 트랜잭션 내부에서는 AFTER_COMMIT 리스너가 아직 동작하지 않아야 한다. - then(asyncEmailService).shouldHaveNoInteractions(); + // BEFORE_COMMIT 리스너는 아직 실행되지 않았으며 SMTP도 호출하지 않는다. + assertThat(jobCount(user)).isZero(); }); // then - assertAfterCommitEmailSent(user); + assertCommittedJob(user); } @Test - @DisplayName("롤백 시 이메일 비동기 전송이 호출되지 않는다") - void sendVerification_rollback_doesNotTriggerListener() { + @DisplayName("롤백 시 인증 토큰과 이메일 작업이 모두 남지 않는다") + void sendVerification_rollback_persistsNeither() { // given User user = saveUser(); @@ -94,7 +96,7 @@ void sendVerification_rollback_doesNotTriggerListener() { }); // then - then(asyncEmailService).shouldHaveNoInteractions(); + assertThat(jobCount(user)).isZero(); assertThat(tokenRepository.findTopByUserIdOrderByCreatedAtDesc(user.getId())).isEmpty(); } @@ -106,15 +108,13 @@ private User saveUser() { return saved; } - private void assertAfterCommitEmailSent(User user) { - ArgumentCaptor emailCaptor = ArgumentCaptor.forClass(String.class); - ArgumentCaptor linkCaptor = ArgumentCaptor.forClass(String.class); - then(asyncEmailService) - .should() - .sendVerificationEmailAsync(emailCaptor.capture(), linkCaptor.capture()); - - assertThat(emailCaptor.getValue()).isEqualTo(user.getEmail()); - assertThat(linkCaptor.getValue()).contains("?token="); + private void assertCommittedJob(User user) { + assertThat(jobCount(user)).isEqualTo(1); assertThat(tokenRepository.findTopByUserIdOrderByCreatedAtDesc(user.getId())).isPresent(); } + + private int jobCount(User user) { + return jdbc.queryForObject( + "SELECT count(*) FROM email_jobs WHERE recipient = ?", Integer.class, user.getEmail()); + } } diff --git a/src/test/java/com/beachcheck/auth/service/AsyncEmailServiceTest.java b/src/test/java/com/beachcheck/auth/service/AsyncEmailServiceTest.java deleted file mode 100644 index 44d78bae..00000000 --- a/src/test/java/com/beachcheck/auth/service/AsyncEmailServiceTest.java +++ /dev/null @@ -1,78 +0,0 @@ -package com.beachcheck.auth.service; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.BDDMockito.given; -import static org.mockito.BDDMockito.then; -import static org.mockito.Mockito.doThrow; - -import io.micrometer.tracing.Tracer; -import org.junit.jupiter.api.DisplayName; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.ArgumentCaptor; -import org.mockito.Captor; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; - -@ExtendWith(MockitoExtension.class) -@DisplayName("비동기 이메일 서비스 단위 테스트") -class AsyncEmailServiceTest { - - private static final String FROM_EMAIL = "no-reply@test.com"; - private static final String USER_EMAIL = "user@test.com"; - private static final String VERIFICATION_LINK = "https://example.com/verify?token=abc"; - private static final String SUBJECT = "이메일 인증"; - private static final String EXPIRATION_TEXT = "30분"; - private static final int EXPIRATION_MINUTES = 30; - - @Mock private RetryingEmailDeliveryService emailDeliveryService; - @Captor private ArgumentCaptor fromCaptor; - @Captor private ArgumentCaptor toCaptor; - @Captor private ArgumentCaptor subjectCaptor; - @Captor private ArgumentCaptor bodyCaptor; - - @Test - @DisplayName("인증 메일 payload를 조립해 전송 위임") - void sendVerificationEmailAsync_success() { - AsyncEmailService service = newService(); - given( - emailDeliveryService.sendVerificationEmail( - anyString(), anyString(), anyString(), anyString())) - .willReturn(RetryingEmailDeliveryService.DeliveryOutcome.SUCCESS); - - service.sendVerificationEmailAsync(USER_EMAIL, VERIFICATION_LINK); - - then(emailDeliveryService) - .should() - .sendVerificationEmail( - fromCaptor.capture(), - toCaptor.capture(), - subjectCaptor.capture(), - bodyCaptor.capture()); - assertThat(fromCaptor.getValue()).isEqualTo(FROM_EMAIL); - assertThat(toCaptor.getValue()).isEqualTo(USER_EMAIL); - assertThat(subjectCaptor.getValue()).isEqualTo(SUBJECT); - assertThat(bodyCaptor.getValue()).contains(VERIFICATION_LINK); - assertThat(bodyCaptor.getValue()).contains(EXPIRATION_TEXT); - } - - @Test - @DisplayName("예상하지 못한 전송 예외는 호출자에게 그대로 전파") - void sendVerificationEmailAsync_unexpectedFailure_propagates() { - AsyncEmailService service = newService(); - doThrow(new IllegalStateException("delivery failed")) - .when(emailDeliveryService) - .sendVerificationEmail(eq(FROM_EMAIL), eq(USER_EMAIL), eq(SUBJECT), anyString()); - - assertThatThrownBy(() -> service.sendVerificationEmailAsync(USER_EMAIL, VERIFICATION_LINK)) - .isInstanceOf(IllegalStateException.class) - .hasMessage("delivery failed"); - } - - private AsyncEmailService newService() { - return new AsyncEmailService(emailDeliveryService, Tracer.NOOP, FROM_EMAIL, EXPIRATION_MINUTES); - } -} diff --git a/src/test/java/com/beachcheck/auth/service/EmailEventListenerTest.java b/src/test/java/com/beachcheck/auth/service/EmailEventListenerTest.java index 7d3aac86..c03d1a3c 100644 --- a/src/test/java/com/beachcheck/auth/service/EmailEventListenerTest.java +++ b/src/test/java/com/beachcheck/auth/service/EmailEventListenerTest.java @@ -15,16 +15,16 @@ class EmailEventListenerTest { private static final String USER_EMAIL = "user@test.com"; private static final String VERIFICATION_LINK = "https://example.com/verify?token=abc"; - @Mock private AsyncEmailService asyncEmailService; + @Mock private EmailJobService emailJobService; @Test @DisplayName("이벤트 payload를 비동기 서비스로 전달") void handleEmailVerificationEvent_forwardsEventPayload() { - EmailEventListener listener = new EmailEventListener(asyncEmailService); + EmailEventListener listener = new EmailEventListener(emailJobService); EmailVerificationEvent event = new EmailVerificationEvent(USER_EMAIL, VERIFICATION_LINK); listener.handleEmailVerificationEvent(event); - then(asyncEmailService).should().sendVerificationEmailAsync(USER_EMAIL, VERIFICATION_LINK); + then(emailJobService).should().enqueue(USER_EMAIL, VERIFICATION_LINK); } } diff --git a/src/test/java/com/beachcheck/auth/service/EmailPayloadCipherTest.java b/src/test/java/com/beachcheck/auth/service/EmailPayloadCipherTest.java new file mode 100644 index 00000000..9f52fad0 --- /dev/null +++ b/src/test/java/com/beachcheck/auth/service/EmailPayloadCipherTest.java @@ -0,0 +1,21 @@ +package com.beachcheck.auth.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import org.junit.jupiter.api.Test; + +class EmailPayloadCipherTest { + @Test + void encryptsWithRandomNonceAndDetectsWrongKey() { + var cipher = new EmailPayloadCipher("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="); + String input = "https://example.com/verify?token=secret"; + String encrypted = cipher.encrypt(input); + assertThat(encrypted).doesNotContain(input).isNotEqualTo(cipher.encrypt(input)); + assertThat(cipher.decrypt(encrypted)).isEqualTo(input); + var other = new EmailPayloadCipher("AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE="); + assertThatThrownBy(() -> other.decrypt(encrypted)).isInstanceOf(IllegalStateException.class); + assertThatThrownBy(() -> new EmailPayloadCipher("")) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/src/test/java/com/beachcheck/auth/service/EmailVerificationServiceTest.java b/src/test/java/com/beachcheck/auth/service/EmailVerificationServiceTest.java index 9b17300e..1aeb7b2d 100644 --- a/src/test/java/com/beachcheck/auth/service/EmailVerificationServiceTest.java +++ b/src/test/java/com/beachcheck/auth/service/EmailVerificationServiceTest.java @@ -52,7 +52,6 @@ class EmailVerificationServiceTest { private static final long EXP_MIN_LOWER_BOUND = EXP_MINUTES - EXP_TOLERANCE_MINUTES; private static final long EXP_MIN_UPPER_BOUND = EXP_MINUTES; - @Mock private AsyncEmailService asyncEmailService; @Mock private EmailVerificationTokenRepository tokenRepository; @Mock private UserRepository userRepository; @Mock private ApplicationEventPublisher eventPublisher; @@ -66,7 +65,6 @@ class EmailVerificationServiceTest { void setUp() { service = new EmailVerificationService( - asyncEmailService, tokenRepository, userRepository, eventPublisher, diff --git a/src/test/java/com/beachcheck/auth/service/RetryingEmailDeliveryServiceTest.java b/src/test/java/com/beachcheck/auth/service/RetryingEmailDeliveryServiceTest.java deleted file mode 100644 index 1b74fabf..00000000 --- a/src/test/java/com/beachcheck/auth/service/RetryingEmailDeliveryServiceTest.java +++ /dev/null @@ -1,62 +0,0 @@ -package com.beachcheck.auth.service; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.mockito.BDDMockito.then; -import static org.mockito.Mockito.doThrow; - -import io.micrometer.tracing.Tracer; -import org.junit.jupiter.api.DisplayName; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; -import org.springframework.mail.MailSendException; - -@ExtendWith(MockitoExtension.class) -@DisplayName("재시도 이메일 전송 서비스 단위 테스트") -class RetryingEmailDeliveryServiceTest { - - private static final String FROM = "from@test.com"; - private static final String TO = "to@test.com"; - private static final String SUBJECT = "subject"; - private static final String BODY = "body"; - - @Mock private EmailSender emailSender; - - @Test - @DisplayName("SMTP 전송을 기존 EmailSender에 위임한다") - void sendVerificationEmail_delegatesToSender() { - RetryingEmailDeliveryService service = newService(); - - assertThat(service.sendVerificationEmail(FROM, TO, SUBJECT, BODY)) - .isEqualTo(RetryingEmailDeliveryService.DeliveryOutcome.SUCCESS); - - then(emailSender).should().send(FROM, TO, SUBJECT, BODY); - } - - @Test - @DisplayName("프록시 밖 단위 호출에서는 SMTP 예외를 그대로 전파한다") - void sendVerificationEmail_failure_propagates() { - RetryingEmailDeliveryService service = newService(); - doThrow(new MailSendException("smtp down")).when(emailSender).send(FROM, TO, SUBJECT, BODY); - - assertThatThrownBy(() -> service.sendVerificationEmail(FROM, TO, SUBJECT, BODY)) - .isInstanceOf(MailSendException.class); - } - - @Test - @DisplayName("재시도 소진 recover는 예외 없이 종료한다") - void recoverFromEmailFailure_completesNormally() { - RetryingEmailDeliveryService service = newService(); - - assertThat( - service.recoverFromEmailFailure( - new MailSendException("smtp down"), FROM, TO, SUBJECT, BODY)) - .isEqualTo(RetryingEmailDeliveryService.DeliveryOutcome.RETRIES_EXHAUSTED); - } - - private RetryingEmailDeliveryService newService() { - return new RetryingEmailDeliveryService(emailSender, Tracer.NOOP); - } -} diff --git a/src/test/java/com/beachcheck/auth/service/SmtpFailureClassifierTest.java b/src/test/java/com/beachcheck/auth/service/SmtpFailureClassifierTest.java new file mode 100644 index 00000000..9cb10f82 --- /dev/null +++ b/src/test/java/com/beachcheck/auth/service/SmtpFailureClassifierTest.java @@ -0,0 +1,59 @@ +package com.beachcheck.auth.service; + +import static org.assertj.core.api.Assertions.assertThat; + +import jakarta.mail.AuthenticationFailedException; +import jakarta.mail.internet.InternetAddress; +import org.eclipse.angus.mail.smtp.SMTPAddressFailedException; +import org.eclipse.angus.mail.smtp.SMTPSendFailedException; +import org.junit.jupiter.api.Test; +import org.springframework.mail.MailAuthenticationException; +import org.springframework.mail.MailParseException; +import org.springframework.mail.MailSendException; + +class SmtpFailureClassifierTest { + private final SmtpFailureClassifier classifier = new SmtpFailureClassifier(); + + @Test + void transientResponsesAndNetworkFailuresAreRetryable() throws Exception { + assertThat( + classifier.retryable( + new MailSendException("network", new java.net.SocketTimeoutException()))) + .isTrue(); + assertThat( + classifier.retryable( + new MailSendException( + "smtp", + new SMTPSendFailedException("DATA", 451, "temporary", null, null, null, null)))) + .isTrue(); + assertThat( + classifier.retryable( + new MailSendException( + "smtp", + new SMTPAddressFailedException( + new InternetAddress("test@example.com"), "RCPT", 450, "temporary")))) + .isTrue(); + } + + @Test + void permanentResponsesAndBadCredentialsFailImmediately() throws Exception { + assertThat(classifier.retryable(new MailAuthenticationException("auth"))).isFalse(); + assertThat(classifier.retryable(new MailParseException("invalid"))).isFalse(); + assertThat( + classifier.retryable( + new MailSendException( + "smtp", + new SMTPSendFailedException("DATA", 554, "permanent", null, null, null, null)))) + .isFalse(); + var smtp = + new SMTPAddressFailedException( + new InternetAddress("test@example.com"), "RCPT", 550, "unknown"); + assertThat(classifier.retryable(new MailSendException(java.util.Map.of("message", smtp)))) + .isFalse(); + assertThat( + classifier.retryable( + new MailSendException("auth", new AuthenticationFailedException()))) + .isFalse(); + assertThat(classifier.retryable(new IllegalStateException("unexpected"))).isFalse(); + } +} diff --git a/src/test/resources/application-test.yml b/src/test/resources/application-test.yml index 15f47825..38908088 100644 --- a/src/test/resources/application-test.yml +++ b/src/test/resources/application-test.yml @@ -51,6 +51,10 @@ spring: # JWT 설정 (테스트용) app: + email: + jobs: + polling-enabled: false + payload-key: AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA= # test-only 32-byte key jwt: secret: test-secret-key-for-testing-only-minimum-256-bits-required-for-hmac-sha256 access-token-expiration: 3600000 # 1시간 From 1237434b6b92f547c88a74090931f93f486d4b01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=B0=95=EC=9E=AC=ED=99=8D?= <105151063+PHJ2000@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:32:27 +0900 Subject: [PATCH 2/3] =?UTF-8?q?fix:=20=EC=9D=B4=EB=A9=94=EC=9D=BC=20?= =?UTF-8?q?=EC=9E=91=EC=97=85=20=EC=8B=A4=ED=96=89=20=EA=B2=A9=EB=A6=AC?= =?UTF-8?q?=EC=99=80=20=EB=B9=84=ED=99=9C=EC=84=B1=ED=99=94=20=ED=98=B8?= =?UTF-8?q?=ED=99=98=EC=84=B1=20=EB=B3=B4=EC=99=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 기존 emailTaskExecutor에서 최대 3개 소비 작업만 실행하고 실행 후 DB 선점 - 메일 비활성화 시 작업 생성·처리·키 검증을 함께 제외해 로컬 Noop 흐름 유지 - 무효 작업 정리 뒤에도 20건 배치 한도 안에서 다음 후보 조회 - 복호화 실패를 SMTP 오류와 분리하고 복구 판단을 위한 암호문 보존 - 실행기 포화·작업 실패·공용 스케줄러 격리·설정·DB 회귀 테스트 13개 추가 검증: 전체 337 tests, assemble, Checkstyle main/test, Spotless, UTF-8, Gitleaks 통과 Related #270 --- README.md | 16 ++- deploy/backend-ec2.env.example | 2 +- docs/LOCAL_DEVELOPMENT.md | 2 +- .../auth/repository/EmailJobStore.java | 18 +-- .../auth/service/EmailEventListener.java | 2 + .../auth/service/EmailJobScheduler.java | 31 ++++- .../auth/service/EmailJobService.java | 2 + .../auth/service/EmailJobWorker.java | 29 +++-- .../auth/service/EmailPayloadCipher.java | 2 + .../EmailJobDisabledIntegrationTest.java | 73 ++++++++++++ .../integration/EmailJobIntegrationTest.java | 81 +++++++++++-- .../EmailJobTracingIntegrationTest.java | 6 +- .../EmailJobTransactionIntegrationTest.java | 4 + .../service/EmailJobConfigurationTest.java | 95 +++++++++++++++ .../auth/service/EmailJobSchedulerTest.java | 111 ++++++++++++++++++ 15 files changed, 441 insertions(+), 33 deletions(-) create mode 100644 src/test/java/com/beachcheck/auth/integration/EmailJobDisabledIntegrationTest.java create mode 100644 src/test/java/com/beachcheck/auth/service/EmailJobConfigurationTest.java create mode 100644 src/test/java/com/beachcheck/auth/service/EmailJobSchedulerTest.java diff --git a/README.md b/README.md index 34903be1..2eb5145a 100644 --- a/README.md +++ b/README.md @@ -213,6 +213,9 @@ Beach_complex/ 인증 토큰과 `email_jobs`는 같은 DB 트랜잭션에서 저장한다. 워커는 시작 시와 기본 1초 주기로 `PENDING`, 재시도 시각이 지난 `RETRY_WAIT`, lease가 지난 `SENDING`을 조회한다. `FOR UPDATE SKIP LOCKED`로 선점·시도 횟수 증가·시도 이력 생성을 커밋한 다음 SMTP를 호출한다. +공용 스케줄러는 실행 요청만 전달하고, 기존 `emailTaskExecutor`에서 인스턴스당 최대 3개의 소비 작업을 +실행한다. 실행기 대기 중에는 DB 작업을 선점하지 않는다. 소비 작업 하나는 무효 작업 정리와 발송을 +합쳐 최대 20건 처리하며, 만료 작업을 정리한 뒤에도 배치 한도 안에서 다음 후보를 계속 조회한다. 완료 기록은 현재 시도 번호가 일치할 때만 허용한다. 상태는 `PENDING → SENDING → SUCCEEDED / RETRY_WAIT / FAILED`이며 `email_job_attempts`에 각 시도를 남긴다. @@ -229,9 +232,14 @@ SMTP가 수락한 뒤 DB 성공 기록 전에 종료되면 발송 여부를 확 조사를 돕지만 수신자의 중복 수신을 제거하지 못한다. `SUCCEEDED`는 SMTP 수락이며 받은편지함 도착이 아니다. 4번째 시도가 불명 상태로 끝나면 추가 발송 없이 `FAILED`/`ATTEMPTS_EXHAUSTED_UNKNOWN`으로 남긴다. -배포 전 모든 인스턴스에 `APP_EMAIL_JOBS_PAYLOAD_KEY`를 동일하게 설정해야 한다. +`APP_MAIL_ENABLED=false`이면 기존 개발용 Noop 동작을 유지한다. 인증 토큰은 생성하지만 +메일 작업을 새로 만들거나 기존 작업을 처리하지 않으며 암호화 키도 요구하지 않는다. +실제 메일을 활성화하는 인스턴스에는 `APP_EMAIL_JOBS_PAYLOAD_KEY`를 동일하게 설정해야 한다. `openssl rand -base64 32`로 생성한 32바이트 키를 Secret으로 보관하며 미설정/잘못된 길이는 시작을 차단한다. -인증 링크는 AES-GCM으로 암호화해 저장하고 종료된 작업에서는 삭제한다. 키를 변경하기 전에는 +인증 링크는 AES-GCM으로 암호화해 저장하고 종료된 작업에서는 삭제한다. +단, 복호화 실패는 SMTP에 도달하지 않은 `FAILED/PAYLOAD_DECRYPT_FAILED`로 구분하고 암호문을 보존한다. +자동 재발송은 하지 않으며 키·데이터를 확인한 뒤 별도 복구 판단이 필요하다. 이 처리 시도도 DB 횟수에 남는다. +키를 변경하기 전에는 미완료 작업을 처리하거나 기존 키로 복호화 후 새 키로 재암호화해야 한다. 수신 주소와 시도 이력은 DB에 남으므로 접근을 제한하며, 이 변경에는 자동 보존기간 삭제 작업을 포함하지 않는다. 주소·링크·제목·본문·SMTP 예외 원문은 로그/Trace에 기록하지 않는다. @@ -239,7 +247,7 @@ SMTP가 수락한 뒤 DB 성공 기록 전에 종료되면 발송 여부를 확 설정은 [배포 환경 예시](deploy/backend-ec2.env.example)를 참고한다. 기본 lease는 300초(최소 60초), SMTP 연결/읽기/쓰기 timeout은 각각 5초다. timeout은 전체 전송 시간의 엄밀한 상한은 아니다. -`APP_EMAIL_JOBS_POLLING_ENABLED=false`는 시작 복구와 주기 실행을 모두 끄며 이미 진행 중인 전송을 -취소하지 않는다. 롤백 전 워커를 정지하고 진행 중 요청을 종료한 뒤 미완료 작업을 확인한다. +`APP_EMAIL_JOBS_POLLING_ENABLED=false`는 작업 생성을 유지하면서 시작 복구와 주기 실행만 끈다. +두 설정 모두 적용 전에 이미 시작된 전송을 취소하지 않는다. 롤백 전 워커를 정지하고 진행 중 요청을 종료한 뒤 미완료 작업을 확인한다. 구버전은 DB 작업을 처리하지 못하며, 기존 메모리 작업을 새 테이블로 자동 이전하지 않는다. 배포 전 구버전 요청을 drain해야 하며 운영 배포와 실제 SMTP 발송 검증은 PR 테스트 범위와 별개다. diff --git a/deploy/backend-ec2.env.example b/deploy/backend-ec2.env.example index 86bde2e7..e9b9fdd4 100644 --- a/deploy/backend-ec2.env.example +++ b/deploy/backend-ec2.env.example @@ -13,7 +13,7 @@ APP_EMAIL_VERIFICATION_BASE_URL=https:///api/auth/verify-email SPRING_MAIL_USERNAME= SPRING_MAIL_PASSWORD= APP_EMAIL_FROM_ADDRESS= -# Required: generate once with `openssl rand -base64 32`; same key on every email worker. +# Required when APP_MAIL_ENABLED=true: generate once with `openssl rand -base64 32`; same key on every email worker. # Preserve this secret across restarts; drain outstanding jobs before rotation. APP_EMAIL_JOBS_PAYLOAD_KEY= APP_EMAIL_JOBS_POLLING_ENABLED=true diff --git a/docs/LOCAL_DEVELOPMENT.md b/docs/LOCAL_DEVELOPMENT.md index f90d4fcc..a72ae6e2 100644 --- a/docs/LOCAL_DEVELOPMENT.md +++ b/docs/LOCAL_DEVELOPMENT.md @@ -79,7 +79,7 @@ curl 'http://localhost:8080/api/beaches?lat=35.1587&lon=129.1599&radiusKm=5' | 연동 | 설정 | | --- | --- | | PostgreSQL | `SPRING_DATASOURCE_URL`, `SPRING_DATASOURCE_USERNAME`, `SPRING_DATASOURCE_PASSWORD` | -| 이메일 | `APP_MAIL_ENABLED=true`, `SPRING_MAIL_USERNAME`, `SPRING_MAIL_PASSWORD`; 메일 상태 점검 시 `MANAGEMENT_HEALTH_MAIL_ENABLED=true` | +| 이메일 | `APP_MAIL_ENABLED=true`, `SPRING_MAIL_USERNAME`, `SPRING_MAIL_PASSWORD`, `APP_EMAIL_JOBS_PAYLOAD_KEY`(32바이트 Base64, 모든 메일 인스턴스에 동일한 키); 메일 상태 점검 시 `MANAGEMENT_HEALTH_MAIL_ENABLED=true` | | 이메일 인증 링크 | `APP_EMAIL_VERIFICATION_BASE_URL` — 기본값 `http://localhost:8080/api/auth/verify-email` | | Firebase FCM | `APP_FIREBASE_ENABLED=true`, `APP_OUTBOX_POLLING_ENABLED=true`, 서비스 계정 설정 | | Firebase 서비스 계정 | `APP_FIREBASE_CREDENTIALS_JSON_BASE64` 또는 `APP_FIREBASE_CREDENTIALS_PATH` | diff --git a/src/main/java/com/beachcheck/auth/repository/EmailJobStore.java b/src/main/java/com/beachcheck/auth/repository/EmailJobStore.java index 31b8a34e..8549b0fa 100644 --- a/src/main/java/com/beachcheck/auth/repository/EmailJobStore.java +++ b/src/main/java/com/beachcheck/auth/repository/EmailJobStore.java @@ -36,7 +36,7 @@ public UUID enqueue(String recipient, String payload, String tokenHash, String t /** 선점·횟수 증가·이력 생성은 한 트랜잭션이며 SMTP 호출 전에 커밋한다. */ @Transactional(propagation = Propagation.REQUIRES_NEW) - public Optional claim() { + public ClaimResult claim() { var candidates = jdbc.query( """ @@ -58,7 +58,7 @@ public Optional claim() { rs.getInt("attempts"), rs.getString("producer_traceparent")), rs.getBoolean("valid_token"))); - if (candidates.isEmpty()) return Optional.empty(); + if (candidates.isEmpty()) return new ClaimResult(Optional.empty(), true); Candidate candidate = candidates.getFirst(); Job old = candidate.job(); jdbc.update( @@ -75,8 +75,8 @@ public Optional claim() { """, old.attempt() >= 4 ? "ATTEMPTS_EXHAUSTED_UNKNOWN" : "TOKEN_INACTIVE", old.id()); - // 다음 poll에서 다음 후보를 조회하며, 실패 행은 다시 선택하지 않는다. - return Optional.empty(); + // 정리한 후보도 이번 배치의 한 건으로 세되, 다음 후보를 계속 조회한다. + return new ClaimResult(Optional.empty(), false); } int attempt = old.attempt() + 1; jdbc.update( @@ -89,8 +89,9 @@ public Optional claim() { leaseSeconds, old.id()); jdbc.update("INSERT INTO email_job_attempts(job_id, attempt) VALUES (?, ?)", old.id(), attempt); - return Optional.of( - new Job(old.id(), old.recipient(), old.payload(), attempt, old.traceparent())); + return new ClaimResult( + Optional.of(new Job(old.id(), old.recipient(), old.payload(), attempt, old.traceparent())), + false); } @Transactional(propagation = Propagation.REQUIRES_NEW) @@ -100,7 +101,7 @@ public boolean finish(Job job, String status, String outcome, long delayMs) { """ UPDATE email_jobs SET status = ?, lease_until = NULL, next_retry_at = CASE WHEN ? = 'RETRY_WAIT' THEN CURRENT_TIMESTAMP + (? * INTERVAL '1 millisecond') ELSE NULL END, - payload = CASE WHEN ? IN ('SUCCEEDED', 'FAILED') THEN NULL ELSE payload END, + payload = CASE WHEN ? IN ('SUCCEEDED', 'FAILED') AND ? <> 'PAYLOAD_DECRYPT_FAILED' THEN NULL ELSE payload END, last_error = CASE WHEN ? = 'SUCCEEDED' THEN NULL ELSE ? END WHERE id = ? AND status = 'SENDING' AND attempts = ? """, @@ -108,6 +109,7 @@ public boolean finish(Job job, String status, String outcome, long delayMs) { status, delayMs, status, + outcome, status, outcome, job.id(), @@ -124,6 +126,8 @@ public boolean finish(Job job, String status, String outcome, long delayMs) { return true; } + public record ClaimResult(Optional job, boolean queueEmpty) {} + public record Job(UUID id, String recipient, String payload, int attempt, String traceparent) {} private record Candidate(Job job, boolean validToken) {} diff --git a/src/main/java/com/beachcheck/auth/service/EmailEventListener.java b/src/main/java/com/beachcheck/auth/service/EmailEventListener.java index 3b94b0be..30769e20 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailEventListener.java +++ b/src/main/java/com/beachcheck/auth/service/EmailEventListener.java @@ -1,10 +1,12 @@ package com.beachcheck.auth.service; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; import org.springframework.transaction.event.TransactionPhase; import org.springframework.transaction.event.TransactionalEventListener; @Component +@ConditionalOnProperty(prefix = "app.mail", name = "enabled", havingValue = "true") public class EmailEventListener { private final EmailJobService emailJobService; diff --git a/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java b/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java index 76c06cf4..21e075b2 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java +++ b/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java @@ -1,7 +1,10 @@ package com.beachcheck.auth.service; +import java.util.concurrent.Executor; +import java.util.concurrent.Semaphore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.context.event.EventListener; @@ -10,25 +13,45 @@ @Component @ConditionalOnProperty( - name = "app.email.jobs.polling-enabled", - havingValue = "true", - matchIfMissing = true) + name = {"app.mail.enabled", "app.email.jobs.polling-enabled"}, + havingValue = "true") public class EmailJobScheduler { private static final Logger log = LoggerFactory.getLogger(EmailJobScheduler.class); + // 기존 emailTaskExecutor의 core pool 크기만큼만 소비 작업을 예약한다. + private static final int WORKERS = 3; + private final Semaphore slots = new Semaphore(WORKERS); private final EmailJobWorker worker; + private final Executor executor; - public EmailJobScheduler(EmailJobWorker worker) { + public EmailJobScheduler( + EmailJobWorker worker, @Qualifier("emailTaskExecutor") Executor executor) { this.worker = worker; + this.executor = executor; } @EventListener(ApplicationReadyEvent.class) @Scheduled(fixedDelayString = "${app.email.jobs.poll-interval-ms:1000}") public void poll() { + for (int i = 0; i < WORKERS && slots.tryAcquire(); i++) { + try { + // 실행기에 대기 중인 동안에는 DB 작업을 선점하지 않는다. + executor.execute(this::runWorker); + } catch (RuntimeException failure) { + slots.release(); + log.error("Email job dispatch failed: errorType={}", failure.getClass().getName()); + return; + } + } + } + + private void runWorker() { try { worker.poll(); } catch (RuntimeException failure) { // DB/SMTP 예외 원문은 주소·인증 링크를 포함할 수 있다. log.error("Email job polling failed: errorType={}", failure.getClass().getName()); + } finally { + slots.release(); } } } diff --git a/src/main/java/com/beachcheck/auth/service/EmailJobService.java b/src/main/java/com/beachcheck/auth/service/EmailJobService.java index 311d390b..d12db0ac 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailJobService.java +++ b/src/main/java/com/beachcheck/auth/service/EmailJobService.java @@ -5,10 +5,12 @@ import io.micrometer.tracing.Span; import io.micrometer.tracing.Tracer; import java.util.UUID; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @Service +@ConditionalOnProperty(prefix = "app.mail", name = "enabled", havingValue = "true") public class EmailJobService { private final EmailJobStore store; private final EmailPayloadCipher cipher; diff --git a/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java b/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java index 4b1a3d91..39678f72 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java +++ b/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java @@ -9,9 +9,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Service; @Service +@ConditionalOnProperty(prefix = "app.mail", name = "enabled", havingValue = "true") public class EmailJobWorker { private static final Logger log = LoggerFactory.getLogger(EmailJobWorker.class); private final EmailJobStore store; @@ -37,11 +39,11 @@ public EmailJobWorker( } public void poll() { - // 직렬로 소량 처리하여 메모리 큐에 선점한 작업을 오래 보관하지 않는다. + // 소비 작업 하나당 정리·발송을 합쳐 최대 20건 처리한다. for (int i = 0; i < 20; i++) { var claimed = store.claim(); - if (claimed.isEmpty()) return; - deliver(claimed.get()); + if (claimed.queueEmpty()) return; + claimed.job().ifPresent(this::deliver); } } @@ -62,6 +64,21 @@ private void deliver(Job job) { } Span work = builder.start().tag("email.operation", "verification"); try (Tracer.SpanInScope ignored = tracer.withSpan(work)) { + String link; + try { + link = cipher.decrypt(job.payload()); + } catch (RuntimeException failure) { + // SMTP에 도달하지 못한 설정/데이터 오류: 자동 재시도하지 않고 암호문을 보존한다. + boolean recorded = store.finish(job, "FAILED", "PAYLOAD_DECRYPT_FAILED", 0); + work.tag("email.delivery.outcome", recorded ? "payload_decrypt_failed" : "stale_claim"); + work.error(new IllegalStateException("Email payload decryption failed")); + log.error( + "Email job={} attempt={} outcome={}", + job.id(), + job.attempt(), + recorded ? "PAYLOAD_DECRYPT_FAILED" : "STALE_CLAIM"); + return; + } String status = "SUCCEEDED"; String outcome = "ACCEPTED"; Span smtp = @@ -71,11 +88,7 @@ private void deliver(Job job) { .tag("email.retry.attempt", String.valueOf(job.attempt())) .start(); try (Tracer.SpanInScope smtpScope = tracer.withSpan(smtp)) { - sender.send( - from, - job.recipient(), - "이메일 인증", - "아래 링크를 클릭하여 이메일을 인증해주세요:\n\n" + cipher.decrypt(job.payload())); + sender.send(from, job.recipient(), "이메일 인증", "아래 링크를 클릭하여 이메일을 인증해주세요:\n\n" + link); smtp.tag("email.delivery.outcome", "success"); } catch (RuntimeException failure) { boolean retry = classifier.retryable(failure); diff --git a/src/main/java/com/beachcheck/auth/service/EmailPayloadCipher.java b/src/main/java/com/beachcheck/auth/service/EmailPayloadCipher.java index c8d88ae8..04a11566 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailPayloadCipher.java +++ b/src/main/java/com/beachcheck/auth/service/EmailPayloadCipher.java @@ -8,10 +8,12 @@ import javax.crypto.spec.GCMParameterSpec; import javax.crypto.spec.SecretKeySpec; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; /** 인증 링크 원문을 DB에 노출하지 않는다. 키는 모든 워커에서 동일하게 유지한다. */ @Component +@ConditionalOnProperty(prefix = "app.mail", name = "enabled", havingValue = "true") public class EmailPayloadCipher { private final byte[] key; private final SecureRandom random = new SecureRandom(); diff --git a/src/test/java/com/beachcheck/auth/integration/EmailJobDisabledIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobDisabledIntegrationTest.java new file mode 100644 index 00000000..6c88ef4e --- /dev/null +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobDisabledIntegrationTest.java @@ -0,0 +1,73 @@ +package com.beachcheck.auth.integration; + +import static com.beachcheck.support.fixture.EmailVerificationTestFixtures.emailUser; +import static com.beachcheck.support.fixture.UniqueTestFixtures.uniqueEmail; +import static org.assertj.core.api.Assertions.assertThat; + +import com.beachcheck.auth.service.EmailJobScheduler; +import com.beachcheck.auth.service.EmailJobWorker; +import com.beachcheck.auth.service.EmailVerificationService; +import com.beachcheck.support.base.IntegrationTest; +import com.beachcheck.user.repository.UserRepository; +import java.util.UUID; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.TestPropertySource; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; + +@TestPropertySource( + properties = { + "app.mail.enabled=false", + "app.email.jobs.polling-enabled=true", + "app.email.jobs.payload-key=" + }) +@Transactional(propagation = Propagation.NOT_SUPPORTED) +class EmailJobDisabledIntegrationTest extends IntegrationTest { + @Autowired private ApplicationContext context; + @Autowired private EmailVerificationService verification; + @Autowired private UserRepository users; + @Autowired private JdbcTemplate jdbc; + + @Test + void disabledMailStartsWithoutKeyAndLeavesExistingJobsUntouched() { + var user = users.save(emailUser(uniqueEmail("mail-disabled"), false)); + UUID existingJob = UUID.randomUUID(); + try { + jdbc.update( + "INSERT INTO email_jobs(id, recipient, payload, token_hash) VALUES (?, ?, ?, ?)", + existingJob, + user.getEmail(), + "retained-ciphertext", + "unused-hash"); + verification.sendVerification(user); + assertThat(context.getBeansOfType(EmailJobScheduler.class)).isEmpty(); + assertThat(context.getBeansOfType(EmailJobWorker.class)).isEmpty(); + assertThat( + jdbc.queryForObject( + "SELECT count(*) FROM email_jobs WHERE recipient = ?", + Integer.class, + user.getEmail())) + .isEqualTo(1); + assertThat( + jdbc.queryForObject( + "SELECT status FROM email_jobs WHERE id = ?", String.class, existingJob)) + .isEqualTo("PENDING"); + assertThat( + jdbc.queryForObject( + "SELECT payload FROM email_jobs WHERE id = ?", String.class, existingJob)) + .isEqualTo("retained-ciphertext"); + assertThat( + jdbc.queryForObject( + "SELECT count(*) FROM email_verification_tokens WHERE user_id = ?", + Integer.class, + user.getId())) + .isEqualTo(1); + } finally { + jdbc.update("DELETE FROM email_jobs WHERE id = ?", existingJob); + users.deleteById(user.getId()); + } + } +} diff --git a/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java index caebb724..ad7bbdec 100644 --- a/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java @@ -38,6 +38,7 @@ import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; +@org.springframework.test.context.TestPropertySource(properties = "app.mail.enabled=true") @Transactional(propagation = Propagation.NOT_SUPPORTED) class EmailJobIntegrationTest extends IntegrationTest { @Autowired private EmailVerificationService verification; @@ -72,7 +73,7 @@ void cleanup() { void pendingSurvivesWorkerRecreationAndSuccessIsNotSentAgain() { var restarted = new EmailJobWorker(store, sender, cipher, new SmtpFailureClassifier(), Tracer.NOOP, ""); - new EmailJobScheduler(restarted).poll(); + new EmailJobScheduler(restarted, Runnable::run).poll(); worker.poll(); assertThat(status()).isEqualTo("SUCCEEDED"); assertThat(attempts()).isEqualTo(1); @@ -85,10 +86,10 @@ void pendingSurvivesWorkerRecreationAndSuccessIsNotSentAgain() { @Test void staleSendingRecoversAndOldWorkerCannotOverwriteNewClaim() { - Job old = store.claim().orElseThrow(); - assertThat(store.claim()).isEmpty(); + Job old = store.claim().job().orElseThrow(); + assertThat(store.claim().queueEmpty()).isTrue(); expireLease(); - Job recovered = store.claim().orElseThrow(); + Job recovered = store.claim().job().orElseThrow(); assertThat(recovered.attempt()).isEqualTo(2); assertThat(store.finish(old, "SUCCEEDED", "ACCEPTED", 0)).isFalse(); assertThat(status()).isEqualTo("SENDING"); @@ -214,7 +215,7 @@ void permanentFailureDoesNotRetry() { @Test void fourthAbandonedClaimNeverGetsFifthAttempt() { for (int attempt = 1; attempt <= 4; attempt++) { - assertThat(store.claim().orElseThrow().attempt()).isEqualTo(attempt); + assertThat(store.claim().job().orElseThrow().attempt()).isEqualTo(attempt); expireLease(); } worker.poll(); @@ -238,7 +239,7 @@ void expiredAndSupersededTokensAreNotDelivered() { @Test void closingAndStartingApplicationContextsRecoversPendingAndStaleSending() { try (var first = restartContext()) { - first.getBean(EmailJobStore.class).claim().orElseThrow(); + first.getBean(EmailJobStore.class).claim().job().orElseThrow(); } expireLease(); verification.sendVerification(user); // 다른 PENDING 작업도 재시작 후 처리 @@ -261,6 +262,9 @@ void closingAndStartingApplicationContextsRecoversPendingAndStaleSending() { private org.springframework.context.annotation.AnnotationConfigApplicationContext restartContext() { var context = new org.springframework.context.annotation.AnnotationConfigApplicationContext(); + org.springframework.test.context.support.TestPropertySourceUtils + .addInlinedPropertiesToEnvironment( + context, "app.mail.enabled=true", "app.email.jobs.polling-enabled=true"); context.register(RestartTransactions.class); context.registerBean( org.springframework.transaction.PlatformTransactionManager.class, @@ -281,7 +285,7 @@ void closingAndStartingApplicationContextsRecoversPendingAndStaleSending() { "")); context.registerBean( EmailJobScheduler.class, - () -> new EmailJobScheduler(context.getBean(EmailJobWorker.class))); + () -> new EmailJobScheduler(context.getBean(EmailJobWorker.class), Runnable::run)); context.refresh(); return context; } @@ -326,6 +330,69 @@ void manualResendInvalidatesOldPendingJobWithoutResettingItsAttempts() { verify(sender, times(1)).send(anyString(), eq(user.getEmail()), anyString(), anyString()); } + @Test + void expiredHeadDoesNotStopDeliveryOfFollowingValidJob() { + jdbc.update( + "UPDATE email_verification_tokens SET expires_at = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE user_id = ?", + user.getId()); + verification.sendVerification(user); + worker.poll(); + assertThat(status()).isEqualTo("FAILED"); + assertThat( + jdbc.queryForObject( + "SELECT count(*) FROM email_jobs WHERE recipient = ? AND status = 'SUCCEEDED'", + Integer.class, + user.getEmail())) + .isEqualTo(1); + verify(sender).send(anyString(), eq(user.getEmail()), anyString(), anyString()); + } + + @Test + void invalidCandidatesStillRespectTwentyItemBatchLimit() { + jdbc.update( + "UPDATE email_verification_tokens SET expires_at = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE user_id = ?", + user.getId()); + jdbc.update( + """ + INSERT INTO email_jobs(id, recipient, payload, token_hash) + SELECT uuid_generate_v4(), recipient, payload, token_hash FROM email_jobs + CROSS JOIN generate_series(1, 24) WHERE id = ? + """, + jobId); + worker.poll(); + assertThat( + jdbc.queryForObject( + "SELECT count(*) FROM email_jobs WHERE status = 'FAILED'", Integer.class)) + .isEqualTo(20); + assertThat( + jdbc.queryForObject( + "SELECT count(*) FROM email_jobs WHERE status = 'PENDING'", Integer.class)) + .isEqualTo(5); + verifyNoInteractions(sender); + } + + @Test + void decryptionFailurePreservesCiphertextWithoutCallingSmtp() { + String encrypted = + jdbc.queryForObject("SELECT payload FROM email_jobs WHERE id = ?", String.class, jobId); + var wrongKey = new EmailPayloadCipher("AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE="); + new EmailJobWorker(store, sender, wrongKey, new SmtpFailureClassifier(), Tracer.NOOP, "") + .poll(); + assertThat(status()).isEqualTo("FAILED"); + assertThat(history()).containsExactly("PAYLOAD_DECRYPT_FAILED"); + assertThat( + jdbc.queryForObject( + "SELECT last_error FROM email_jobs WHERE id = ?", String.class, jobId)) + .isEqualTo("PAYLOAD_DECRYPT_FAILED"); + String preserved = + jdbc.queryForObject("SELECT payload FROM email_jobs WHERE id = ?", String.class, jobId); + assertThat(preserved).isEqualTo(encrypted); + assertThat(cipher.decrypt(preserved)).contains("?token="); + worker.poll(); + verifyNoInteractions(sender); + assertThat(attempts()).isEqualTo(1); + } + private void expireLease() { jdbc.update( "UPDATE email_jobs SET lease_until = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE id = ?", diff --git a/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java index cbb3a30d..d375911f 100644 --- a/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java @@ -77,7 +77,11 @@ void recoveredJobLinksProducerAndDoesNotLeakSensitiveData() { 3, traceparent); org.mockito.Mockito.when(store.claim()) - .thenReturn(java.util.Optional.of(job), java.util.Optional.empty()); + .thenReturn( + new com.beachcheck.auth.repository.EmailJobStore.ClaimResult( + java.util.Optional.of(job), false), + new com.beachcheck.auth.repository.EmailJobStore.ClaimResult( + java.util.Optional.empty(), true)); org.mockito.Mockito.when( store.finish( org.mockito.ArgumentMatchers.any(), diff --git a/src/test/java/com/beachcheck/auth/integration/EmailJobTransactionIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobTransactionIntegrationTest.java index ba49c414..1893325f 100644 --- a/src/test/java/com/beachcheck/auth/integration/EmailJobTransactionIntegrationTest.java +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobTransactionIntegrationTest.java @@ -23,6 +23,7 @@ import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionTemplate; +@org.springframework.test.context.TestPropertySource(properties = "app.mail.enabled=true") @Transactional(propagation = Propagation.NOT_SUPPORTED) class EmailJobTransactionIntegrationTest extends IntegrationTest { @@ -31,6 +32,9 @@ class EmailJobTransactionIntegrationTest extends IntegrationTest { @Autowired private EmailVerificationTokenRepository tokenRepository; @Autowired private PlatformTransactionManager transactionManager; + @org.springframework.boot.test.mock.mockito.MockBean + private com.beachcheck.auth.service.EmailSender sender; + @Autowired private JdbcTemplate jdbc; private final List createdUserIds = new ArrayList<>(); diff --git a/src/test/java/com/beachcheck/auth/service/EmailJobConfigurationTest.java b/src/test/java/com/beachcheck/auth/service/EmailJobConfigurationTest.java new file mode 100644 index 00000000..95482681 --- /dev/null +++ b/src/test/java/com/beachcheck/auth/service/EmailJobConfigurationTest.java @@ -0,0 +1,95 @@ +package com.beachcheck.auth.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +import com.beachcheck.auth.repository.EmailJobStore; +import io.micrometer.tracing.Tracer; +import java.util.concurrent.Executor; +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +class EmailJobConfigurationTest { + private final ApplicationContextRunner runner = + new ApplicationContextRunner() + .withBean(EmailJobStore.class, () -> mock(EmailJobStore.class)) + .withBean(Tracer.class, () -> Tracer.NOOP) + .withBean(EmailSender.class, () -> mock(EmailSender.class)) + .withBean("emailTaskExecutor", Executor.class, () -> Runnable::run) + .withUserConfiguration( + EmailPayloadCipher.class, + EmailJobService.class, + EmailJobWorker.class, + EmailEventListener.class, + EmailJobScheduler.class, + SmtpFailureClassifier.class) + .withPropertyValues("app.email.jobs.polling-enabled=true"); + + @Test + void mailDisabledDoesNotRequireKeyOrRegisterAnyJobProducerOrConsumer() { + runner + .withPropertyValues("app.mail.enabled=false") + .run( + context -> { + assertThat(context).hasNotFailed(); + assertThat(context) + .doesNotHaveBean(EmailPayloadCipher.class) + .doesNotHaveBean(EmailEventListener.class) + .doesNotHaveBean(EmailJobService.class) + .doesNotHaveBean(EmailJobWorker.class) + .doesNotHaveBean(EmailJobScheduler.class); + }); + } + + @Test + void missingMailSettingAlsoPreservesNoopMode() { + runner.run( + context -> { + assertThat(context).hasNotFailed(); + assertThat(context) + .doesNotHaveBean(EmailEventListener.class) + .doesNotHaveBean(EmailJobWorker.class) + .doesNotHaveBean(EmailPayloadCipher.class); + }); + } + + @Test + void actualMailStillFailsFastWithoutKey() { + runner + .withPropertyValues("app.mail.enabled=true") + .run( + context -> { + assertThat(context).hasFailed(); + assertThat(context.getStartupFailure()) + .hasRootCauseInstanceOf(IllegalArgumentException.class) + .rootCause() + .hasMessageContaining("APP_EMAIL_JOBS_PAYLOAD_KEY"); + }); + } + + @Test + void pollingDisabledStillAllowsDurableEnqueue() { + runner + .withPropertyValues( + "app.mail.enabled=true", + "app.email.jobs.polling-enabled=false", + "app.email.jobs.payload-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=") + .run( + context -> { + assertThat(context) + .hasNotFailed() + .hasSingleBean(EmailJobService.class) + .hasSingleBean(EmailEventListener.class) + .doesNotHaveBean(EmailJobScheduler.class); + }); + } + + @Test + void mailEnabledRegistersSchedulerUsingExistingExecutor() { + runner + .withPropertyValues( + "app.mail.enabled=true", + "app.email.jobs.payload-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=") + .run(context -> assertThat(context).hasNotFailed().hasSingleBean(EmailJobScheduler.class)); + } +} diff --git a/src/test/java/com/beachcheck/auth/service/EmailJobSchedulerTest.java b/src/test/java/com/beachcheck/auth/service/EmailJobSchedulerTest.java new file mode 100644 index 00000000..4790fe47 --- /dev/null +++ b/src/test/java/com/beachcheck/auth/service/EmailJobSchedulerTest.java @@ -0,0 +1,111 @@ +package com.beachcheck.auth.service; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; + +import com.beachcheck.global.config.AsyncConfig; +import io.micrometer.tracing.Tracer; +import java.time.Instant; +import java.util.ArrayDeque; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +class EmailJobSchedulerTest { + @Test + void repeatedTicksReserveAtMostThreeConsumersWithoutClaimingBeforeExecution() { + var worker = mock(EmailJobWorker.class); + var queued = new ArrayDeque(); + var scheduler = new EmailJobScheduler(worker, queued::add); + scheduler.poll(); + scheduler.poll(); + assertThat(queued).hasSize(3); + verifyNoInteractions(worker); + queued.remove().run(); + verify(worker).poll(); + scheduler.poll(); + assertThat(queued).hasSize(3); + } + + @Test + void rejectedDispatchReleasesSlotForNextTick() { + var worker = mock(EmailJobWorker.class); + var queued = new ArrayDeque(); + var reject = new AtomicBoolean(true); + var scheduler = + new EmailJobScheduler( + worker, + task -> { + if (reject.getAndSet(false)) throw new RejectedExecutionException("full"); + queued.add(task); + }); + scheduler.poll(); + verifyNoInteractions(worker); + scheduler.poll(); + assertThat(queued).hasSize(3); + } + + @Test + void failedConsumersReleaseTheirSlots() { + var worker = mock(EmailJobWorker.class); + doThrow(new IllegalStateException("failure")).when(worker).poll(); + var queued = new ArrayDeque(); + var scheduler = new EmailJobScheduler(worker, queued::add); + scheduler.poll(); + for (int i = 0; i < 3; i++) queued.remove().run(); + scheduler.poll(); + assertThat(queued).hasSize(3); + verify(worker, times(3)).poll(); + } + + @Test + void slowConsumersUseExistingExecutorWithoutBlockingUnrelatedScheduledWork() throws Exception { + var worker = mock(EmailJobWorker.class); + var entered = new CountDownLatch(3); + var release = new CountDownLatch(1); + var finished = new CountDownLatch(3); + doAnswer( + call -> { + entered.countDown(); + try { + assertThat(release.await(5, TimeUnit.SECONDS)).isTrue(); + } finally { + finished.countDown(); + } + return null; + }) + .when(worker) + .poll(); + var executor = (ThreadPoolTaskExecutor) new AsyncConfig().emailTaskExecutor(Tracer.NOOP); + executor.setWaitForTasksToCompleteOnShutdown(true); + executor.setAwaitTerminationSeconds(5); + var sharedScheduler = new ThreadPoolTaskScheduler(); + sharedScheduler.setPoolSize(1); + sharedScheduler.initialize(); + try { + var scheduler = new EmailJobScheduler(worker, executor); + sharedScheduler.schedule(scheduler::poll, Instant.now()).get(2, TimeUnit.SECONDS); + assertThat(entered.await(2, TimeUnit.SECONDS)).isTrue(); + var unrelated = new CountDownLatch(1); + sharedScheduler.schedule(unrelated::countDown, Instant.now()); + assertThat(unrelated.await(2, TimeUnit.SECONDS)).isTrue(); + scheduler.poll(); + assertThat(executor.getThreadPoolExecutor().getQueue()).isEmpty(); + verify(worker, times(3)).poll(); + } finally { + release.countDown(); + assertThat(finished.await(5, TimeUnit.SECONDS)).isTrue(); + sharedScheduler.shutdown(); + executor.shutdown(); + } + } +} From ef294334776724b68bd6e9373549d0acb54628de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=B0=95=EC=9E=AC=ED=99=8D?= <105151063+PHJ2000@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:59:24 +0900 Subject: [PATCH 3/3] =?UTF-8?q?perf:=20=EC=9D=B4=EB=A9=94=EC=9D=BC=20?= =?UTF-8?q?=EC=9C=A0=ED=9C=B4=20=EC=A1=B0=ED=9A=8C=EC=99=80=20=EC=A4=91?= =?UTF-8?q?=EB=B3=B5=20=ED=85=8C=EC=8A=A4=ED=8A=B8=20=EC=A0=95=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 주기마다 소비 작업 하나를 요청하고 실제 작업 발견 시에만 최대 3개로 확장 - 기존 실행기·동시 실행 제한·DB 선점·20건 배치 한도 유지 - 미사용 Trace 발송 fixture와 RetryConfig, 중복 비활성화 설정 검증 제거 - 유휴 조회·작업 발견 알림·슬롯 반환 검증을 현재 실행 흐름에 맞게 조정 검증: 전체 338 tests, assemble, Checkstyle main/test, Spotless, UTF-8, Gitleaks 통과 Related #270 --- README.md | 4 +- .../auth/service/EmailJobScheduler.java | 31 ++++++---- .../auth/service/EmailJobWorker.java | 12 ++++ .../integration/EmailJobIntegrationTest.java | 11 ++++ .../EmailJobTracingIntegrationTest.java | 34 +---------- .../service/EmailJobConfigurationTest.java | 16 ------ .../auth/service/EmailJobSchedulerTest.java | 57 +++++++++++++------ 7 files changed, 89 insertions(+), 76 deletions(-) diff --git a/README.md b/README.md index 2eb5145a..12eb532d 100644 --- a/README.md +++ b/README.md @@ -214,7 +214,9 @@ Beach_complex/ `PENDING`, 재시도 시각이 지난 `RETRY_WAIT`, lease가 지난 `SENDING`을 조회한다. `FOR UPDATE SKIP LOCKED`로 선점·시도 횟수 증가·시도 이력 생성을 커밋한 다음 SMTP를 호출한다. 공용 스케줄러는 실행 요청만 전달하고, 기존 `emailTaskExecutor`에서 인스턴스당 최대 3개의 소비 작업을 -실행한다. 실행기 대기 중에는 DB 작업을 선점하지 않는다. 소비 작업 하나는 무효 작업 정리와 발송을 +실행한다. 주기 실행은 소비 작업 하나만 요청하며, 실제 발송할 작업을 발견했을 때 나머지 슬롯을 채운다. +따라서 유휴 상태에서는 매 주기마다 소비 작업 3개가 빈 큐를 조회하지 않는다. +실행기 대기 중에는 DB 작업을 선점하지 않는다. 소비 작업 하나는 무효 작업 정리와 발송을 합쳐 최대 20건 처리하며, 만료 작업을 정리한 뒤에도 배치 한도 안에서 다음 후보를 계속 조회한다. 완료 기록은 현재 시도 번호가 일치할 때만 허용한다. 상태는 `PENDING → SENDING → SUCCEEDED / RETRY_WAIT / FAILED`이며 `email_job_attempts`에 각 시도를 남긴다. diff --git a/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java b/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java index 21e075b2..509f77e0 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java +++ b/src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java @@ -32,21 +32,32 @@ public EmailJobScheduler( @EventListener(ApplicationReadyEvent.class) @Scheduled(fixedDelayString = "${app.email.jobs.poll-interval-ms:1000}") public void poll() { - for (int i = 0; i < WORKERS && slots.tryAcquire(); i++) { - try { - // 실행기에 대기 중인 동안에는 DB 작업을 선점하지 않는다. - executor.execute(this::runWorker); - } catch (RuntimeException failure) { - slots.release(); - log.error("Email job dispatch failed: errorType={}", failure.getClass().getName()); - return; - } + submitWorker(); + } + + private boolean submitWorker() { + if (!slots.tryAcquire()) return false; + try { + // 실행기에 대기 중인 동안에는 DB 작업을 선점하지 않는다. + executor.execute(this::runWorker); + return true; + } catch (RuntimeException failure) { + slots.release(); + log.error("Email job dispatch failed: errorType={}", failure.getClass().getName()); + return false; + } + } + + private void expandWorkers() { + // 실제 발송할 작업을 발견했을 때만 나머지 슬롯을 채운다. + for (int i = 1; i < WORKERS; i++) { + if (!submitWorker()) return; } } private void runWorker() { try { - worker.poll(); + worker.poll(this::expandWorkers); } catch (RuntimeException failure) { // DB/SMTP 예외 원문은 주소·인증 링크를 포함할 수 있다. log.error("Email job polling failed: errorType={}", failure.getClass().getName()); diff --git a/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java b/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java index 39678f72..c214e84d 100644 --- a/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java +++ b/src/main/java/com/beachcheck/auth/service/EmailJobWorker.java @@ -39,10 +39,22 @@ public EmailJobWorker( } public void poll() { + poll( + () -> { + // 직접 호출은 추가 소비 작업을 예약하지 않는다. + }); + } + + public void poll(Runnable onFirstJob) { + boolean notified = false; // 소비 작업 하나당 정리·발송을 합쳐 최대 20건 처리한다. for (int i = 0; i < 20; i++) { var claimed = store.claim(); if (claimed.queueEmpty()) return; + if (claimed.job().isPresent() && !notified) { + notified = true; + onFirstJob.run(); + } claimed.job().ifPresent(this::deliver); } } diff --git a/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java index ad7bbdec..a6a58777 100644 --- a/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobIntegrationTest.java @@ -393,6 +393,17 @@ void decryptionFailurePreservesCiphertextWithoutCallingSmtp() { assertThat(attempts()).isEqualTo(1); } + @Test + void workerReportsOnlyFirstSendableJobToScheduler() { + verification.sendVerification(user); + Runnable onFirstJob = org.mockito.Mockito.mock(Runnable.class); + worker.poll(onFirstJob); + verify(onFirstJob).run(); + verify(sender, times(2)).send(anyString(), eq(user.getEmail()), anyString(), anyString()); + worker.poll(onFirstJob); + verify(onFirstJob).run(); // 빈 큐를 다시 조회해도 확장 알림을 보내지 않는다. + } + private void expireLease() { jdbc.update( "UPDATE email_jobs SET lease_until = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE id = ?", diff --git a/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java b/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java index d375911f..3437cc3c 100644 --- a/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java +++ b/src/test/java/com/beachcheck/auth/integration/EmailJobTracingIntegrationTest.java @@ -4,9 +4,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import com.beachcheck.auth.service.EmailSender; import com.beachcheck.global.config.AsyncConfig; -import com.beachcheck.global.config.RetryConfig; import com.beachcheck.global.logging.MdcTaskDecorator; import com.beachcheck.support.tracing.RecordingSpanExporter; import io.micrometer.tracing.Span; @@ -15,7 +13,6 @@ import io.opentelemetry.sdk.trace.SdkTracerProvider; import io.opentelemetry.sdk.trace.data.SpanData; import java.util.List; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -51,8 +48,7 @@ class EmailJobTracingIntegrationTest { .OpenTelemetryAutoConfiguration.class, OpenTelemetryAutoConfiguration.class, MicrometerTracingAutoConfiguration.class)) - .withUserConfiguration( - AsyncConfig.class, RetryConfig.class, EmailTraceTestConfiguration.class) + .withUserConfiguration(AsyncConfig.class, EmailTraceTestConfiguration.class) .withPropertyValues("management.tracing.sampling.probability=1.0"); @Test @@ -256,33 +252,5 @@ static class EmailTraceTestConfiguration { RecordingSpanExporter recordingSpanExporter() { return new RecordingSpanExporter(); } - - @Bean - RecordingEmailSender emailSender() { - return new RecordingEmailSender(); - } - } - - static class RecordingEmailSender implements EmailSender { - - private volatile CountDownLatch latch = new CountDownLatch(0); - - void prepare() { - latch = new CountDownLatch(1); - } - - @Override - public void send(String from, String to, String subject, String body) { - latch.countDown(); - } - - boolean await() { - try { - return latch.await(3, TimeUnit.SECONDS); - } catch (InterruptedException exception) { - Thread.currentThread().interrupt(); - return false; - } - } } } diff --git a/src/test/java/com/beachcheck/auth/service/EmailJobConfigurationTest.java b/src/test/java/com/beachcheck/auth/service/EmailJobConfigurationTest.java index 95482681..c9cd4430 100644 --- a/src/test/java/com/beachcheck/auth/service/EmailJobConfigurationTest.java +++ b/src/test/java/com/beachcheck/auth/service/EmailJobConfigurationTest.java @@ -25,22 +25,6 @@ class EmailJobConfigurationTest { SmtpFailureClassifier.class) .withPropertyValues("app.email.jobs.polling-enabled=true"); - @Test - void mailDisabledDoesNotRequireKeyOrRegisterAnyJobProducerOrConsumer() { - runner - .withPropertyValues("app.mail.enabled=false") - .run( - context -> { - assertThat(context).hasNotFailed(); - assertThat(context) - .doesNotHaveBean(EmailPayloadCipher.class) - .doesNotHaveBean(EmailEventListener.class) - .doesNotHaveBean(EmailJobService.class) - .doesNotHaveBean(EmailJobWorker.class) - .doesNotHaveBean(EmailJobScheduler.class); - }); - } - @Test void missingMailSettingAlsoPreservesNoopMode() { runner.run( diff --git a/src/test/java/com/beachcheck/auth/service/EmailJobSchedulerTest.java b/src/test/java/com/beachcheck/auth/service/EmailJobSchedulerTest.java index 4790fe47..14542aac 100644 --- a/src/test/java/com/beachcheck/auth/service/EmailJobSchedulerTest.java +++ b/src/test/java/com/beachcheck/auth/service/EmailJobSchedulerTest.java @@ -1,6 +1,7 @@ package com.beachcheck.auth.service; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -15,56 +16,79 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; class EmailJobSchedulerTest { @Test - void repeatedTicksReserveAtMostThreeConsumersWithoutClaimingBeforeExecution() { + void emptyQueueRunsOneConsumerPerTickWithoutSpeculativeFanout() { var worker = mock(EmailJobWorker.class); var queued = new ArrayDeque(); var scheduler = new EmailJobScheduler(worker, queued::add); scheduler.poll(); - scheduler.poll(); - assertThat(queued).hasSize(3); + assertThat(queued).hasSize(1); verifyNoInteractions(worker); + queued.remove().run(); // 작업 발견 알림이 없으면 소비 작업을 늘리지 않는다. + assertThat(queued).isEmpty(); + verify(worker).poll(any(Runnable.class)); + scheduler.poll(); + assertThat(queued).hasSize(1); + } + + @Test + void findingWorkExpandsConsumersButRepeatedTicksRemainBounded() { + var worker = mock(EmailJobWorker.class); + var queued = new ArrayDeque(); + doAnswer( + call -> { + call.getArgument(0, Runnable.class).run(); + return null; + }) + .when(worker) + .poll(any(Runnable.class)); + var scheduler = new EmailJobScheduler(worker, queued::add); + scheduler.poll(); queued.remove().run(); - verify(worker).poll(); + assertThat(queued).hasSize(2); + scheduler.poll(); scheduler.poll(); assertThat(queued).hasSize(3); + verify(worker).poll(any(Runnable.class)); } @Test void rejectedDispatchReleasesSlotForNextTick() { var worker = mock(EmailJobWorker.class); var queued = new ArrayDeque(); - var reject = new AtomicBoolean(true); + var rejections = new AtomicInteger(3); var scheduler = new EmailJobScheduler( worker, task -> { - if (reject.getAndSet(false)) throw new RejectedExecutionException("full"); + if (rejections.getAndDecrement() > 0) throw new RejectedExecutionException("full"); queued.add(task); }); - scheduler.poll(); + for (int i = 0; i < 3; i++) scheduler.poll(); verifyNoInteractions(worker); scheduler.poll(); - assertThat(queued).hasSize(3); + assertThat(queued).hasSize(1); } @Test void failedConsumersReleaseTheirSlots() { var worker = mock(EmailJobWorker.class); - doThrow(new IllegalStateException("failure")).when(worker).poll(); + doThrow(new IllegalStateException("failure")).when(worker).poll(any(Runnable.class)); var queued = new ArrayDeque(); var scheduler = new EmailJobScheduler(worker, queued::add); + for (int i = 0; i < 3; i++) { + scheduler.poll(); + queued.remove().run(); + } scheduler.poll(); - for (int i = 0; i < 3; i++) queued.remove().run(); - scheduler.poll(); - assertThat(queued).hasSize(3); - verify(worker, times(3)).poll(); + assertThat(queued).hasSize(1); + verify(worker, times(3)).poll(any(Runnable.class)); } @Test @@ -75,6 +99,7 @@ void slowConsumersUseExistingExecutorWithoutBlockingUnrelatedScheduledWork() thr var finished = new CountDownLatch(3); doAnswer( call -> { + call.getArgument(0, Runnable.class).run(); entered.countDown(); try { assertThat(release.await(5, TimeUnit.SECONDS)).isTrue(); @@ -84,7 +109,7 @@ void slowConsumersUseExistingExecutorWithoutBlockingUnrelatedScheduledWork() thr return null; }) .when(worker) - .poll(); + .poll(any(Runnable.class)); var executor = (ThreadPoolTaskExecutor) new AsyncConfig().emailTaskExecutor(Tracer.NOOP); executor.setWaitForTasksToCompleteOnShutdown(true); executor.setAwaitTerminationSeconds(5); @@ -100,7 +125,7 @@ void slowConsumersUseExistingExecutorWithoutBlockingUnrelatedScheduledWork() thr assertThat(unrelated.await(2, TimeUnit.SECONDS)).isTrue(); scheduler.poll(); assertThat(executor.getThreadPoolExecutor().getQueue()).isEmpty(); - verify(worker, times(3)).poll(); + verify(worker, times(3)).poll(any(Runnable.class)); } finally { release.countDown(); assertThat(finished.await(5, TimeUnit.SECONDS)).isTrue();