Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -16,16 +16,16 @@
*
* <p>Contract(Input): datasource-micrometer가 생성한 QueryContext
*
* <p>Contract(Output): ERROR span에는 원래 예외 클래스명과 민감정보가 제거된 exception event만 기록
* <p>Contract(Output): 오류가 발생한 JDBC query span에는 원래 예외 클래스명과 민감정보가 제거된 exception event만 기록
*/
@Component
@Order(MicrometerTracingAutoConfiguration.DEFAULT_TRACING_OBSERVATION_HANDLER_ORDER - 1000)
public class SanitizingQueryTracingObservationHandler extends QueryTracingObservationHandler {
public class JdbcQueryErrorSanitizingObservationHandler extends QueryTracingObservationHandler {

private static final String ERROR_TYPE = "error.type";
private static final String SANITIZED_ERROR_MESSAGE = "JDBC 쿼리 실행 실패";

public SanitizingQueryTracingObservationHandler(Tracer tracer) {
public JdbcQueryErrorSanitizingObservationHandler(Tracer tracer) {
super(tracer);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import com.beachcheck.outbox.service.OutboxPublisher;
import com.beachcheck.outbox.tracing.OutboxTracing;
import com.google.firebase.messaging.FirebaseMessaging;
import io.micrometer.tracing.Tracer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
Expand Down Expand Up @@ -33,9 +34,10 @@ public class OutboxFirebaseConfig {
public OutboxEventDispatcher outboxEventDispatcher(
OutboxEventRepository outboxEventRepository,
NotificationRepository notificationRepository,
FirebaseMessaging firebaseMessaging) {
FirebaseMessaging firebaseMessaging,
Tracer tracer) {
return new OutboxEventDispatcher(
outboxEventRepository, notificationRepository, firebaseMessaging);
outboxEventRepository, notificationRepository, firebaseMessaging, tracer);
}

@Bean
Expand Down
230 changes: 193 additions & 37 deletions src/main/java/com/beachcheck/outbox/service/OutboxEventDispatcher.java
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,18 @@
import com.google.firebase.messaging.FirebaseMessagingException;
import com.google.firebase.messaging.Message;
import com.google.firebase.messaging.MessagingErrorCode;
import io.micrometer.tracing.Span;
import io.micrometer.tracing.Tracer;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

/**
* Why: OutboxPublisher의 self-invocation 문제로 @Transactional(REQUIRES_NEW)가 프록시를 우회하는 것을 방지하기 위해 별도
Expand All @@ -31,14 +37,17 @@ public class OutboxEventDispatcher {
private final OutboxEventRepository outboxEventRepository;
private final NotificationRepository notificationRepository;
private final FirebaseMessaging firebaseMessaging;
private final Tracer tracer;

public OutboxEventDispatcher(
OutboxEventRepository outboxEventRepository,
NotificationRepository notificationRepository,
FirebaseMessaging firebaseMessaging) {
FirebaseMessaging firebaseMessaging,
Tracer tracer) {
this.outboxEventRepository = outboxEventRepository;
this.notificationRepository = notificationRepository;
this.firebaseMessaging = firebaseMessaging;
this.tracer = tracer;
}

/**
Expand All @@ -48,51 +57,98 @@ public OutboxEventDispatcher(
*/
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void dispatch(OutboxEvent event) {
Span span = startSpan("outbox dispatch");
AtomicBoolean outcomeRecorded = new AtomicBoolean(false);
boolean transactionSynchronizationActive =
TransactionSynchronizationManager.isSynchronizationActive();
if (transactionSynchronizationActive) {
registerCompletionCallback(span, outcomeRecorded, event);
}
try (Tracer.SpanInScope ignored = tracer.withSpan(span)) {
dispatchEvent(event, span, outcomeRecorded);
} catch (RuntimeException | Error exception) {
outcomeRecorded.set(true);
span.tag("outbox.item.outcome", "error");
span.tag("outbox.failure.class", exception.getClass().getName());
span.error(new IllegalStateException("Outbox dispatch 실패"));
throw exception;
} finally {
if (!transactionSynchronizationActive) {
span.end();
}
}
}

private void dispatchEvent(OutboxEvent event, Span dispatchSpan, AtomicBoolean outcomeRecorded) {
// 1. Notification 조회
Notification notification =
notificationRepository
.findById(event.getNotificationId())
.orElseThrow(() -> new IllegalArgumentException("Notification을 찾을 수 없습니다"));
traceChild(
"outbox notification lookup",
() ->
notificationRepository
.findById(event.getNotificationId())
.orElseThrow(() -> new IllegalArgumentException("Notification을 찾을 수 없습니다")));

// 2. 멱등성: 이미 SENT 상태면 OutboxEvent만 SENT로 전이하고 스킵
if (notification.getStatus() == NotificationStatus.SENT) {
event.markAsSent();
outboxEventRepository.save(event);
log.info(
"Outbox 이벤트 멱등 스킵 (이미 발송됨)",
kv("outboxEventId", event.getId()),
kv("notificationId", event.getNotificationId()),
kv("outboxEventType", event.getEventType()));
traceStateUpdate(
() -> {
event.markAsSent();
outboxEventRepository.save(event);
});
recordAfterCommit(
() -> {
outcomeRecorded.set(true);
dispatchSpan.tag("outbox.item.outcome", "skipped");
dispatchSpan.tag("outbox.item.skip.reason", "already_sent");
log.info(
"Outbox 이벤트 멱등 스킵 (이미 발송됨)",
kv("outboxEventId", event.getId()),
kv("notificationId", event.getNotificationId()),
kv("outboxEventType", event.getEventType()));
});
return;
}

// 3. FCM 전송
try {
Message message = notification.toFcmMessage();
firebaseMessaging.send(message);
sendFcm(message, event);

traceStateUpdate(
() -> {
// 4. Notification 상태 업데이트
notification.setStatus(NotificationStatus.SENT);
notification.setSentAt(Instant.now());
notificationRepository.save(notification);

// 4. Notification 상태 업데이트
notification.setStatus(NotificationStatus.SENT);
notification.setSentAt(Instant.now());
notificationRepository.save(notification);

// 5. OutboxEvent 상태 업데이트
event.markAsSent();
outboxEventRepository.save(event);

log.info(
"Outbox 이벤트 발송 성공",
kv("outboxEventId", event.getId()),
kv("notificationId", event.getNotificationId()),
kv("outboxEventType", event.getEventType()),
kv("retryCount", event.getRetryCount()));
// 5. OutboxEvent 상태 업데이트
event.markAsSent();
outboxEventRepository.save(event);
});
recordAfterCommit(
() -> {
outcomeRecorded.set(true);
dispatchSpan.tag("outbox.item.outcome", "success");
log.info(
"Outbox 이벤트 발송 성공",
kv("outboxEventId", event.getId()),
kv("notificationId", event.getNotificationId()),
kv("outboxEventType", event.getEventType()),
kv("retryCount", event.getRetryCount()));
});
} catch (FirebaseMessagingException e) {
// Exponential Backoff 재시도 로직
if (isPermanentFcmError(e)) {
event.markAsFailedPermanent();
notification.markAsFailed("errorCode: " + e.getMessagingErrorCode());
notificationRepository.save(notification);
outboxEventRepository.save(event);
traceStateUpdate(
() -> {
event.markAsFailedPermanent();
notification.markAsFailed("errorCode: " + e.getMessagingErrorCode());
notificationRepository.save(notification);
outboxEventRepository.save(event);
});
tagFailureOutcome(
dispatchSpan, outcomeRecorded, "permanent_failure", event.getRetryCount());
log.warn(
"Outbox 이벤트 발송 영구 실패",
kv("outboxEventId", event.getId()),
Expand All @@ -103,19 +159,29 @@ public void dispatch(OutboxEvent event) {
Duration backoff = Duration.ofSeconds(1L << event.getRetryCount()); // 1s, 2s, 4s

if (event.getRetryCount() >= 3) { // 최대 재시도 횟수 초과 시 영구 실패로 전이
event.markAsFailedPermanent();
notification.markAsFailed("errorCode: " + e.getMessagingErrorCode());
notificationRepository.save(notification);
outboxEventRepository.save(event);
traceStateUpdate(
() -> {
event.markAsFailedPermanent();
notification.markAsFailed("errorCode: " + e.getMessagingErrorCode());
notificationRepository.save(notification);
outboxEventRepository.save(event);
});
tagFailureOutcome(
dispatchSpan, outcomeRecorded, "permanent_failure", event.getRetryCount());
log.warn(
"Outbox 이벤트 발송 영구 실패 (최대 재시도 초과)",
kv("outboxEventId", event.getId()),
kv("notificationId", event.getNotificationId()),
kv("retryCount", event.getRetryCount()),
kv("messagingErrorCode", e.getMessagingErrorCode()));
} else {
event.markAsFailedRetriable(backoff);
outboxEventRepository.save(event);
traceStateUpdate(
() -> {
event.markAsFailedRetriable(backoff);
outboxEventRepository.save(event);
});
tagFailureOutcome(
dispatchSpan, outcomeRecorded, "retryable_failure", event.getRetryCount());
log.warn(
"Outbox 이벤트 발송 실패 (재시도 예정)",
kv("outboxEventId", event.getId()),
Expand All @@ -126,6 +192,96 @@ public void dispatch(OutboxEvent event) {
}
}

private void sendFcm(Message message, OutboxEvent event) throws FirebaseMessagingException {
Span span = startClientSpan("outbox fcm send");
try (Tracer.SpanInScope ignored = tracer.withSpan(span)) {
firebaseMessaging.send(message);
span.tag("outbox.fcm.outcome", "success");
} catch (FirebaseMessagingException exception) {
String outcome =
isPermanentFcmError(exception) || event.getRetryCount() >= 3
? "permanent_failure"
: "retryable_failure";
span.tag("outbox.fcm.outcome", outcome);
span.error(new IllegalStateException("FCM 전송 실패"));
throw exception;
} finally {
span.end();
}
}

private void traceStateUpdate(Runnable action) {
traceChild(
"outbox state update",
() -> {
action.run();
return null;
});
}

private <T> T traceChild(String name, Supplier<T> action) {
Span span = startSpan(name);
try (Tracer.SpanInScope ignored = tracer.withSpan(span)) {
return action.get();
} finally {
span.end();
}
}

private Span startSpan(String name) {
return tracer.nextSpan().name(name).start();
}

private Span startClientSpan(String name) {
return tracer
.spanBuilder()
.setParent(tracer.currentSpan().context())
.name(name)
.kind(Span.Kind.CLIENT)
.start();
}

private void tagFailureOutcome(
Span span, AtomicBoolean outcomeRecorded, String outcome, int retryCount) {
outcomeRecorded.set(true);
span.tag("outbox.item.outcome", outcome);
span.tag("outbox.retry.count", retryCount);
}

private void recordAfterCommit(Runnable action) {
if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.registerSynchronization(
new TransactionSynchronization() {
@Override
public void afterCommit() {
action.run();
}
});
return;
}
action.run();
}

private void registerCompletionCallback(
Span span, AtomicBoolean outcomeRecorded, OutboxEvent event) {
TransactionSynchronizationManager.registerSynchronization(
new TransactionSynchronization() {
@Override
public void afterCompletion(int status) {
if (status != TransactionSynchronization.STATUS_COMMITTED
&& outcomeRecorded.compareAndSet(false, true)) {
span.tag("outbox.item.outcome", "error");
span.error(new IllegalStateException("Outbox dispatch 커밋 실패"));
log.warn(
"Outbox 이벤트 상태 저장 커밋 실패",
kv("outboxEventId", event.getId()),
kv("notificationId", event.getNotificationId()));
}
span.end();
}
});
}

private boolean isPermanentFcmError(FirebaseMessagingException e) {
MessagingErrorCode errorCode = e.getMessagingErrorCode();
return errorCode == MessagingErrorCode.UNREGISTERED
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ void autoConfiguration_wrapsDataSourceAndEnablesQuerySpansOnly() {
.hasSize(1)
.allSatisfy(
(name, handler) ->
assertThat(handler).isInstanceOf(SanitizingQueryTracingObservationHandler.class));
assertThat(handler).isInstanceOf(JdbcQueryErrorSanitizingObservationHandler.class));
assertThat(context.getBeansOfType(OpenTelemetryMeterObservationHandler.class)).isEmpty();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import com.google.firebase.FirebaseApp;
import com.google.firebase.FirebaseOptions;
import com.google.firebase.messaging.FirebaseMessaging;
import io.micrometer.tracing.Tracer;
import java.io.InputStream;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
Expand All @@ -31,7 +32,8 @@ class OutboxFirebaseConditionContextTest {
.withPropertyValues("app.outbox.polling.enabled=true", "app.outbox.polling.batch-size=10")
.withBean(OutboxEventRepository.class, () -> mock(OutboxEventRepository.class))
.withBean(NotificationRepository.class, () -> mock(NotificationRepository.class))
.withBean(OutboxTracing.class, () -> mock(OutboxTracing.class));
.withBean(OutboxTracing.class, () -> mock(OutboxTracing.class))
.withBean(Tracer.class, () -> mock(Tracer.class));

@Test
@DisplayName("Firebase가 비활성화되면 Outbox 관련 빈 없이도 컨텍스트가 정상 기동된다")
Expand Down
Loading
Loading