이벤트·Outbox·멱등 소비
세션 완료 transaction 안에서 broker에 메시지를 바로 보내면 두 저장소를 원자적으로 묶을 수 없습니다. DB commit은 성공하고 broker 전송은 실패하거나, broker는 받았는데 DB가 rollback되는 부분 성공이 생깁니다.
Outbox는 세션 상태와 “보낼 이벤트” 행을 같은 PostgreSQL transaction에 저장합니다. 별도 relay가 commit된 행만 broker로 전달하므로 원본 변경과 외부 전송 사이의 틈을 재시도 가능한 DB 상태로 바꿉니다.
이중 쓰기는 부분 성공을 만든다
Outbox가 exactly-once를 만들어 주는 것은 아닙니다. relay가 broker 전송에는 성공하고 publishedAt 기록 전에 죽으면 같은 행을 다시 보냅니다. 따라서 소비자는 eventId와 업무 반영을 자기 DB transaction 안에서 함께 기록해야 합니다.
세션 완료와 Outbox 저장을 한 트랜잭션에 묶는다
V1에서 만든 study_session 옆에 V2 outbox 테이블을 추가합니다. 애플리케이션이 ddl-auto=validate를 사용하므로 Entity만 추가하고 migration을 빠뜨리면 시작 단계에서 실패합니다.
CREATE TABLE outbox_event (
id UUID PRIMARY KEY,
event_id UUID NOT NULL UNIQUE,
event_type VARCHAR(80) NOT NULL,
payload TEXT NOT NULL,
occurred_at TIMESTAMPTZ NOT NULL,
published_at TIMESTAMPTZ
);
CREATE INDEX idx_outbox_event_pending
ON outbox_event (occurred_at, id)
WHERE published_at IS NULL;package com.andongmin.studylog.outbox;
import java.time.Instant;
import java.util.UUID;
import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
@Entity
@Table(name = "outbox_event")
public class OutboxEvent {
@Id
private UUID id;
@Column(nullable = false, unique = true)
private UUID eventId;
@Column(nullable = false, length = 80)
private String eventType;
@Column(nullable = false, columnDefinition = "text")
private String payload;
@Column(nullable = false)
private Instant occurredAt;
private Instant publishedAt;
protected OutboxEvent() {
}
public OutboxEvent(UUID eventId, String eventType, String payload, Instant occurredAt) {
this.id = UUID.randomUUID();
this.eventId = eventId;
this.eventType = eventType;
this.payload = payload;
this.occurredAt = occurredAt;
}
public UUID eventId() { return eventId; }
public String eventType() { return eventType; }
public String payload() { return payload; }
public void markPublished(Instant time) { this.publishedAt = time; }
}package com.andongmin.studylog.outbox;
import java.util.UUID;
import org.springframework.data.jpa.repository.JpaRepository;
public interface OutboxRepository extends JpaRepository<OutboxEvent, UUID> {
}package com.andongmin.studylog.outbox;
import java.time.Clock;
import java.time.Instant;
import java.util.UUID;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.andongmin.studylog.session.StudySessionJpaRepository;
@Service
public class SessionCompletionService {
private final StudySessionJpaRepository sessions;
private final OutboxRepository outbox;
private final Clock clock;
public SessionCompletionService(StudySessionJpaRepository sessions,
OutboxRepository outbox, Clock clock) {
this.sessions = sessions;
this.outbox = outbox;
this.clock = clock;
}
@Transactional
public UUID complete(UUID sessionId) {
var session = sessions.findById(sessionId).orElseThrow();
session.complete();
UUID eventId = UUID.randomUUID();
String payload = "{\"sessionId\":\"" + sessionId + "\"}";
outbox.save(new OutboxEvent(eventId, "SessionCompleted", payload,
Instant.now(clock)));
return eventId;
}
}SessionCompletionService.complete가 예외로 끝나면 dirty checking으로 바뀐 세션과 outbox insert가 함께 rollback됩니다. 반대로 commit되면 eventId가 안정된 재전송 키가 되고 payload는 broker 종류와 무관한 전달 자료가 됩니다.
여기까지의 코드는 transaction 쪽만 완성합니다. relay와 broker adapter, 소비자 테이블은 아직 제공하지 않으므로 메시지가 실제 전송되거나 한 번만 반영됐다고 주장하지 않습니다.
relay는 published_at IS NULL 행을 작은 batch로 잠그고 broker 확인 응답을 받은 뒤에만 markPublished해야 합니다. 여러 relay replica가 같은 행을 동시에 선택하지 않도록 FOR UPDATE SKIP LOCKED나 동등한 임대 정책도 필요합니다.
transaction 원자성과 전달 멱등성을 따로 검증한다
이 문서에는 SessionCompletionServiceIntegrationTest 파일이나 broker를 제공하지 않습니다. 존재하지 않는 테스트를 실행하지 않고 먼저 migration과 Java 타입이 컴파일되는지 확인합니다. 통합 테스트를 추가할 때는 outbox 저장 실패가 세션 완료도 rollback하는지와 정상 commit이 두 행을 함께 남기는지를 실제 PostgreSQL에서 검증합니다.
./mvnw -q -DskipTests compile컴파일은 원자성을 증명하지 않습니다. 실제 검증에서는 `study_session` 변경과 `outbox_event` 삽입이 함께 commit되거나 함께 rollback되는지 확인하고, relay·consumer 테스트는 제어 가능한 가짜 broker나 Testcontainers broker를 별도로 준비한 뒤 수행합니다.at-least-once 전달을 기본 가정으로 둔다
broker 재시도와 relay 재시작은 같은 이벤트를 다시 보낼 수 있습니다. 소비 결과와 eventId 기록을 같은 트랜잭션으로 묶습니다.
소비자 DB에는 consumed_event(event_id PRIMARY KEY, consumed_at) 같은 처리 기록을 둡니다. eventId insert와 보고서 증가를 한 transaction으로 묶으면 중복 insert가 충돌한 재전송은 업무 값을 다시 바꾸지 않습니다.
비동기 전달은 원격 장애를 요청 transaction 밖으로 옮깁니다. 동기 조회가 반드시 필요한 경로에는 다음 절처럼 timeout과 제한된 실패 예산을 명시해야 합니다.