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 @@ -97,6 +97,9 @@ public void claim(LocalDateTime claimedLeaseUntil) {
this.leaseUntil = truncateToMicros(claimedLeaseUntil);
}

/**
* 재선점 후 도착한 이전 워커의 응답이 현재 작업 상태를 덮지 못하도록 lease를 비교한다.
*/
public boolean isClaimedWith(LocalDateTime claimedLeaseUntil) {
return status == DeliveryStatus.PROCESSING
&& Objects.equals(leaseUntil, truncateToMicros(claimedLeaseUntil));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,10 @@

public interface NotificationDeliveryRepository extends JpaRepository<NotificationDelivery, Long> {

/**
* 동시에 들어온 크롤러 콜백이 같은 event-token 작업을 생성하더라도
* unique 충돌로 트랜잭션을 롤백하지 않고 기존 작업을 유지한다.
*/
@Modifying(flushAutomatically = true)
@Query(value = """
INSERT INTO notification_deliveries (
Expand Down Expand Up @@ -42,6 +46,10 @@ int insertPendingIfAbsent(
@Param("createdAt") LocalDateTime createdAt
);

/**
* 다른 워커가 잠근 행은 기다리지 않고 건너뛰며, lease가 만료된 작업은 다시 선점한다.
* 반환된 행의 잠금은 호출한 claimBatch() 트랜잭션이 끝날 때까지 유지된다.
*/
@Query(value = """
SELECT delivery.id
FROM notification_deliveries delivery
Expand Down Expand Up @@ -77,6 +85,9 @@ List<Long> findEligibleIdsForUpdate(
""")
List<NotificationDelivery> findAllForDispatchByIdIn(@Param("ids") Collection<Long> ids);

/**
* 결과 반영과 재선점이 교차하지 않도록 행을 잠가 lease 확인과 상태 변경을 직렬화한다.
*/
@Lock(LockModeType.PESSIMISTIC_WRITE)
@Query("""
SELECT delivery
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ public class InternalAlertService {

@Transactional
public void createAlert(InternalAlertCommand command) {
// 동일 구독의 중복 콜백을 직렬화해 Summary 검사와 Outbox 생성을 한 임계 구역에서 처리한다.
Subscription subscription = subscriptionRepository.findByIdForUpdate(command.subscriptionId())
.orElseThrow(() -> BaseException.type(CommonErrorCode.ENTITY_NOT_FOUND));

Expand Down Expand Up @@ -61,6 +62,7 @@ public void createAlert(InternalAlertCommand command) {
return;
}

// 구독이 달라도 같은 URL의 같은 게시글은 하나의 이벤트로 식별해 토큰별 중복 작업을 막는다.
String eventId = CryptoUtils.sha256(
subscription.getUrl().getId() + ":" + command.sitePostId());
LocalDateTime now = LocalDateTime.now();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,11 @@
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
* 선점 트랜잭션을 끝낸 뒤 FCM을 호출하고, 결과는 별도 트랜잭션으로 저장한다.
* 외부 호출 중에는 DB 잠금과 커넥션을 점유하지 않는다. FCM 성공 직후 프로세스가 종료되면
* lease 만료 후 재발송될 수 있으므로 전체 전달 보장은 exactly-once가 아닌 at-least-once다.
*/
@Slf4j
@Component
@RequiredArgsConstructor
Expand All @@ -35,6 +40,7 @@ public int dispatchPendingDeliveries() {
return 0;
}

// 하나의 Multicast는 payload를 공유하므로 메시지 내용과 eventId가 모두 같은 작업만 묶는다.
Map<MessageKey, List<ClaimedDelivery>> groups = claimed.stream()
.collect(Collectors.groupingBy(
delivery -> new MessageKey(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ Duration delay(
if (retryAfter == null || retryAfter.isNegative()) {
return backoff;
}
// 서버 백오프보다 FCM Retry-After가 길면 공급자가 요구한 최소 대기 시간을 우선한다.
return retryAfter.compareTo(backoff) > 0 ? retryAfter : backoff;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
@RequiredArgsConstructor
public class NotificationDeliveryService {

// 최초 전송 1회와 최대 3회의 재시도를 합한 횟수다.
public static final int MAX_ATTEMPTS = 4;
public static final int MAX_BATCH_SIZE = 500;
private static final Duration LEASE_DURATION = Duration.ofMinutes(5);
Expand All @@ -46,6 +47,7 @@ public List<ClaimedDelivery> claimBatch(int requestedBatchSize) {
public List<ClaimedDelivery> claimBatch(int requestedBatchSize, LocalDateTime requestedAt) {
int batchSize = Math.max(1, Math.min(requestedBatchSize, MAX_BATCH_SIZE));
LocalDateTime now = truncateToMicros(requestedAt);
// 후보 행 잠금부터 PROCESSING 전환까지 한 트랜잭션으로 묶어 선점을 원자적으로 만든다.
List<Long> eligibleIds = deliveryRepository.findEligibleIdsForUpdate(now, batchSize);
if (eligibleIds.isEmpty()) {
return List.of();
Expand Down Expand Up @@ -118,6 +120,7 @@ public void applyResults(Collection<DeliveryResult> results, LocalDateTime compl
(first, ignored) -> first,
LinkedHashMap::new
));
// 행 잠금 아래에서 lease를 검사해야 늦은 결과와 만료 작업의 재선점이 서로 덮어쓰지 않는다.
List<NotificationDelivery> processingDeliveries = deliveryRepository.findAllByStatusAndIdIn(
DeliveryStatus.PROCESSING,
resultByDeliveryId.keySet()
Expand Down Expand Up @@ -145,6 +148,7 @@ private void applyResult(NotificationDelivery delivery, DeliveryResult result, L
String errorCode = normalizedErrorCode(result.errorCode());
if (result.unregistered()) {
delivery.markFailed(errorCode);
// 발송 중 토큰이 갱신됐을 수 있으므로 실제 시도한 토큰과 현재 값이 같을 때만 끈다.
int deactivatedCount = fcmRepository.deactivateIfTokenMatches(
delivery.getFcmToken().getId(),
result.attemptedToken(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
@RequiredArgsConstructor
public class SummaryCleanupScheduler {

// Summary의 cascade 삭제로 미완료 발송 작업이 사라지지 않도록 정리 대상에서 제외한다.
private static final Set<DeliveryStatus> IN_FLIGHT_DELIVERY_STATUSES = Set.of(
DeliveryStatus.PENDING,
DeliveryStatus.PROCESSING,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,10 @@ public interface FCMRepository extends JpaRepository<FCM_Token, Long> {

FCM_Token findByUserId(Long userId);

/**
* 발송 당시 토큰과 현재 토큰이 같을 때만 비활성화해,
* 늦은 UNREGISTERED 응답이 이미 갱신된 토큰을 끄지 않게 한다.
*/
@Modifying
@Query("""
UPDATE FCM_Token token
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -118,13 +118,15 @@ public List<FcmSendResult> sendMulticast(

ApnsConfig apnsConfig = ApnsConfig.builder()
.putHeader("apns-priority", "10")
// APNs에 대기 중인 동일 이벤트의 병합을 요청하며 이미 표시된 알림까지 제거하지는 않는다.
.putHeader("apns-collapse-id", eventId)
.setAps(Aps.builder().setSound("default").setBadge(1).build())
.build();

MulticastMessage message = MulticastMessage.builder()
.setNotification(notification)
.setApnsConfig(apnsConfig)
// 클라이언트가 eventId를 기준으로 중복 표시를 방지할 수 있도록 함께 전달한다.
.putData("eventId", eventId)
.addAllTokens(targets.stream().map(FcmTarget::token).toList())
.build();
Expand Down Expand Up @@ -158,6 +160,7 @@ private List<FcmSendResult> mapResponse(BatchResponse response, List<FcmTarget>
List<SendResponse> responses = response == null ? null : response.getResponses();
List<FcmSendResult> results = new ArrayList<>(targets.size());

// Admin SDK가 입력 토큰과 응답 순서를 보존하므로 같은 index의 발송 건에 결과를 대응한다.
for (int index = 0; index < targets.size(); index++) {
FcmTarget target = targets.get(index);
if (responses == null || index >= responses.size() || responses.get(index) == null) {
Expand Down Expand Up @@ -237,6 +240,7 @@ private FcmSendResult failureResult(FcmTarget target, FirebaseMessagingException
);
}

/** Retry-After의 delta-seconds와 HTTP-date 형식을 모두 지연 시간으로 변환한다. */
private Duration retryAfterOf(FirebaseMessagingException exception) {
if (exception.getHttpResponse() == null) {
return null;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@ public class FcmTokenLifecycleService {

private final FCMRepository fcmRepository;

/**
* 트랜잭션 없이 실행되는 직접 발송 경로에서도 토큰 무효화만 독립적으로 커밋한다.
*/
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void deactivateAllIfTokenMatches(Collection<FcmTarget> attemptedTokens) {
if (attemptedTokens.isEmpty()) {
Expand Down
Loading