[Feat] 아웃박스 트랜잭션 패턴 적용

[youyeon11]

relavant issue number

• [Feat] 아웃박스 트랜잭션 패턴 적용 #29

어떤 이유로 MR를 하셨나요?

• feature 병합(feature issue #를 남겨주세요)
• 버그 수정(아래에 issue #를 남겨주세요)
• 코드 개선
• 기타(아래에 자세한 내용 기입해주세요)

세부 내용

• gp_ledgers 테이블 변경
  • 기존에 사용되지 않던 원장 데이터를 기록할 수 있도록 gp_ledgers 테이블 구조를 수정

• ChallengeTransactionHelper

        // idempotency guard
        Long assignmentId = challengeAssignment.getId();
        try {
            GpLedger ledger = GpLedger.challengeCompleteReward(userCode, assignmentId, challengeAssignment.getOrigin());
            // 즉시 저장
            gpLedgerRepository.saveAndFlush(ledger);

            // 보상 이벤트 발행
            ChallengeCompletedEvent event = ChallengeCompletedEvent.fromChallengeAssignment(challengeAssignment);
            eventPublisher.publishEvent(event);
        } catch (DataIntegrityViolationException e) {
            LogUtils.info("이미 보상이 처리된 챌린지 완료입니다. assignmentId={}, userCode={}",
                    challengeAssignment.getId(), userCode);
        }

• 멱등성(idempotency) 보장
  • assignmentId + userCode + origin 기반의 unique key를 생성하여 중복 보상 처리를 방지
• 트랜잭션 일관성 유지
  • gp_ledger 저장을 동일한 트랜잭션 내에서 수행하도록 하여 롤백이 가능하도록 구성
• 중복 요청 처리
  • 동일한 챌린지 완료 요청이 들어올 경우 DataIntegrityViolationException을 통해 중복 처리를 감지하고 이벤트 발행을 방지

• save가 아니라 saveAndFlush인 이유
• DataIntegrityViolationException을 그 자리에서 catch하기 위해서 saveAndFlush를 사용
• save() 쓰면 INSERT는 트랜잭션 커밋 시점까지 지연(flush)되기 때문에, unique constraint 위반 예외가 try-catch 블록 바깥에서 터짐

• save() 사용 시 흐름

try {
    gpLedgerRepository.save(ledger);  // 1차 캐시에만 저장, DB 반영되지 않음
    eventPublisher.publishEvent(event);
} catch (DataIntegrityViolationException e) {
    // 여기서 예외가 잡히지 않음
}

// 트랜잭션 커밋 시점에 flush → 이 시점에서 예외 발생

• saveAndFlush() 사용 시 흐름

try {
    gpLedgerRepository.saveAndFlush(ledger);  // 즉시 INSERT
    eventPublisher.publishEvent(event);
} catch (DataIntegrityViolationException e) {
    // 여기서 정상적으로 예외 처리 가능
}

• 즉시 DB에 INSERT를 수행하여 unique constraint 검증이 바로 발생
• 중복일 경우 예외가 즉시 발생하므로 이벤트 발행을 차단

• 세마포어 획득 중 무한대기에 대하여 방지하도록 수정

try {
    databaseSemaphore.acquire();
    acquired = true;
    joinSet = userTransactionHelper.executeRegisterSurvey(requestDto, userCode);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    throw SemaphoreAcquisitionException.of("DB 접근 대기 중 인터럽트 발생", e);
} finally {
    if (acquired) databaseSemaphore.release();
}

• Outbox 패턴 도입
• OutboxPollingServiceImpl

@Transactional
public int publishPendingOutboxEvents() {
    List<OutboxEvent> outboxEvents = outboxEventRepository.findPendingForUpdate(batchSize);
    if (outboxEvents.isEmpty()) {
        return 0;
    }

    int sentCount = 0;
    int failedCount = 0;

    for (OutboxEvent outboxEvent : outboxEvents) {
        try {
            kafkaProducer.send(KafkaTopics.OUTBOX_EVENT, outboxEvent.getPayload());
            outboxEvent.markSent();
            sentCount++;
        } catch (Exception e) {
            outboxEvent.markFailed(trimErrorMessage(e.getMessage()));
            failedCount++;
            LogUtils.error("Failed to publish outbox event: outboxId={}", outboxEvent.getId(), e);
        }
    }

    LogUtils.info("Outbox polling completed: total={}, sent={}, failed={}",
            outboxEvents.size(), sentCount, failedCount);
    return sentCount;
}

• pending 상태의 이벤트를 조회하여 Kafka로 발행
• 발행 성공 시에 sent, 실패하면 failed로 업데이트
• 실패 이벤트는 재처리 대상

• OutboxPublishJob

@Component
@RequiredArgsConstructor
public class OutboxPublishJob {

    private final OutboxPollingService outboxPollingService;

    @Scheduled(
            fixedDelayString = "${outbox.poll.fixed-delay-ms:5000}",
            initialDelayString = "${outbox.poll.initial-delay-ms:5000}"
    )
    public void publishPendingOutboxEvents() {
        int publishedCount = outboxPollingService.publishPendingOutboxEvents();
        if (publishedCount > 0) {
            LogUtils.info("Published outbox events: count={}", publishedCount);
        }
    }
}

• 스케줄러를 통해 주기적으로 Outbox 이벤트를 폴링
• pending 재처리를 통해 이벤트 유실 최소화