[feat] 인증 이메일 작업 영속화 및 재시작·다중 서버 재시도 복구

README.md

@@ -207,3 +207,49 @@ Beach_complex/
 ## 라이선스
 
 [MIT License](./LICENSE)
+
+### 인증 이메일 작업 복구
+
+인증 토큰과 `email_jobs`는 같은 DB 트랜잭션에서 저장한다. 워커는 시작 시와 기본 1초 주기로
+`PENDING`, 재시도 시각이 지난 `RETRY_WAIT`, lease가 지난 `SENDING`을 조회한다.
+`FOR UPDATE SKIP LOCKED`로 선점·시도 횟수 증가·시도 이력 생성을 커밋한 다음 SMTP를 호출한다.
+공용 스케줄러는 실행 요청만 전달하고, 기존 `emailTaskExecutor`에서 인스턴스당 최대 3개의 소비 작업을
+실행한다. 주기 실행은 소비 작업 하나만 요청하며, 실제 발송할 작업을 발견했을 때 나머지 슬롯을 채운다.
+따라서 유휴 상태에서는 매 주기마다 소비 작업 3개가 빈 큐를 조회하지 않는다.
+실행기 대기 중에는 DB 작업을 선점하지 않는다. 소비 작업 하나는 무효 작업 정리와 발송을
+합쳐 최대 20건 처리하며, 만료 작업을 정리한 뒤에도 배치 한도 안에서 다음 후보를 계속 조회한다.
+완료 기록은 현재 시도 번호가 일치할 때만 허용한다. 상태는
+`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_MAIL_ENABLED=false`이면 기존 개발용 Noop 동작을 유지한다. 인증 토큰은 생성하지만
+메일 작업을 새로 만들거나 기존 작업을 처리하지 않으며 암호화 키도 요구하지 않는다.
+실제 메일을 활성화하는 인스턴스에는 `APP_EMAIL_JOBS_PAYLOAD_KEY`를 동일하게 설정해야 한다.
+`openssl rand -base64 32`로 생성한 32바이트 키를 Secret으로 보관하며 미설정/잘못된 길이는 시작을 차단한다.
+인증 링크는 AES-GCM으로 암호화해 저장하고 종료된 작업에서는 삭제한다.
+단, 복호화 실패는 SMTP에 도달하지 않은 `FAILED/PAYLOAD_DECRYPT_FAILED`로 구분하고 암호문을 보존한다.
+자동 재발송은 하지 않으며 키·데이터를 확인한 뒤 별도 복구 판단이 필요하다. 이 처리 시도도 DB 횟수에 남는다.
+키를 변경하기 전에는
+미완료 작업을 처리하거나 기존 키로 복호화 후 새 키로 재암호화해야 한다. 수신 주소와 시도 이력은 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 테스트 범위와 별개다.

deploy/backend-ec2.env.example

@@ -13,3 +13,9 @@ APP_EMAIL_VERIFICATION_BASE_URL=https://<your-domain>/api/auth/verify-email
 SPRING_MAIL_USERNAME=
 SPRING_MAIL_PASSWORD=
 APP_EMAIL_FROM_ADDRESS=
+# 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
+APP_EMAIL_JOBS_POLL_INTERVAL_MS=1000
+APP_EMAIL_JOBS_LEASE_SECONDS=300

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` |

src/main/java/com/beachcheck/auth/repository/EmailJobStore.java

@@ -0,0 +1,134 @@
+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 ClaimResult 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 new ClaimResult(Optional.empty(), true);
+    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());
+      // 정리한 후보도 이번 배치의 한 건으로 세되, 다음 후보를 계속 조회한다.
+      return new ClaimResult(Optional.empty(), false);
+    }
+    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 new ClaimResult(
+        Optional.of(new Job(old.id(), old.recipient(), old.payload(), attempt, old.traceparent())),
+        false);
+  }
+
+  @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') 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 = ?
+        """,
+            status,
+            status,
+            delayMs,
+            status,
+            outcome,
+            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 ClaimResult(Optional<Job> job, boolean queueEmpty) {}
+
+  public record Job(UUID id, String recipient, String payload, int attempt, String traceparent) {}
+
+  private record Candidate(Job job, boolean validToken) {}
+}

src/main/java/com/beachcheck/auth/service/EmailEventListener.java

@@ -1,20 +1,22 @@
 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 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());
   }
 }

src/main/java/com/beachcheck/auth/service/EmailJobScheduler.java

@@ -0,0 +1,68 @@
+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;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+@Component
+@ConditionalOnProperty(
+    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, @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() {
+    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(this::expandWorkers);
+    } catch (RuntimeException failure) {
+      // DB/SMTP 예외 원문은 주소·인증 링크를 포함할 수 있다.
+      log.error("Email job polling failed: errorType={}", failure.getClass().getName());
+    } finally {
+      slots.release();
+    }
+  }
+}