DongHyeonka
d646c2f12f
messaging-superpowers-package 설계서/계획서 기반 구현. registry를 19 → 43 leaf로 확장하고 src/messaging 아래 24개 leaf를 등록. - core-api: M1 publish/consume + M2 batch·delayed·pause-resume - policy/transport-spi: 재시도 결정, DLQ orchestration, admission control, lifecycle - kafka·rabbit(Stable): contiguous commit, confirm/return 상관, 배치, 보안 설정 - pulsar·nats(Experimental): 기본 비활성, live 인증 없음을 코드로 기록 - outbox/inbox/claim-check: 트랜잭션 결합, lease, 무결성 검증 - admin: plan → approve → execute를 타입으로 강제 - 문서 9종, infra compose 7종, JMH 벤치마크 3종 검증: 아키텍처 게이트 3종 통과, 24개 leaf 전부 check 통과, messaging 테스트 604개 통과/0 실패. 미완: 계획서가 요구한 실 브로커 IT 40개 중 7개만 작성. Rabbit 13 / Outbox 6 / Inbox 4 / NATS·Pulsar·Share 5 / testkit 2 / starter·admin 3, 그리고 TLS·ACL 2개가 남음.
96 lines
3.5 KiB
Markdown
96 lines
3.5 KiB
Markdown
# Outbox · Inbox
|
|
|
|
## 두 패턴이 각각 무엇을 해결하는가
|
|
|
|
| 패턴 | 해결하는 문제 | 해결하지 않는 문제 |
|
|
|---|---|---|
|
|
| Transactional Outbox | DB commit과 publish 사이의 창(窓) | 중복 |
|
|
| Inbox | 중복 delivery의 side effect | 유실 |
|
|
|
|
**둘 다 필요하다.** Outbox만으로는 exactly-once가 되지 않는다.
|
|
|
|
## Outbox
|
|
|
|
business transaction과 **같은 transaction**에서 row를 쓴다. 둘 다 commit되거나 둘 다 안 된다.
|
|
|
|
```sql
|
|
BEGIN;
|
|
UPDATE orders SET status = 'PLACED' WHERE id = ?;
|
|
INSERT INTO messaging_outbox (message_id, destination, ...) VALUES (?, ?, ...);
|
|
COMMIT;
|
|
```
|
|
|
|
### relay
|
|
|
|
```text
|
|
leaseBatch(100, 30s) -- lease로 다중 relay 인스턴스 안전
|
|
→ publish (messageId 그대로)
|
|
→ CONFIRMED → markPublished
|
|
→ AMBIGUOUS → markAmbiguous (같은 messageId로 재시도 가능)
|
|
→ REJECTED → markFailed
|
|
```
|
|
|
|
### 핵심 규칙: ambiguous는 같은 messageId로 재시도
|
|
|
|
새 id를 발급하면 "전달됐을 수도 있는 메시지"가 "확실히 두 번째인 메시지"가 되어
|
|
downstream의 어떤 중복 제거도 복구할 수 없다.
|
|
failed로 표시하면 broker가 이미 가지고 있을 수 있는 메시지를 잃는다.
|
|
|
|
`message_id`를 primary key로 둔 것도 같은 이유다. 어떤 코드 경로도 실수로 새 id를 붙일 수 없다.
|
|
|
|
### lease
|
|
|
|
```text
|
|
status IN ('PENDING','AMBIGUOUS') AND (lease_expires_at IS NULL OR lease_expires_at <= now)
|
|
```
|
|
|
|
partial index `ix_messaging_outbox_claimable`이 이 쿼리를 backlog 크기에 비례하게 유지한다.
|
|
PUBLISHED row는 retention job이 지울 때까지 쌓이기 때문이다.
|
|
|
|
## Inbox
|
|
|
|
reservation과 side effect가 **같은 transaction**이어야 한다.
|
|
|
|
```java
|
|
transactions.inTransaction(() -> {
|
|
if (!inbox.reserve(messageId, consumerId, now)) {
|
|
return InboxOutcome.duplicate(); // 이미 처리됨
|
|
}
|
|
return InboxOutcome.processed(sideEffect.get());
|
|
});
|
|
```
|
|
|
|
별도 transaction으로 예약하면 Inbox가 닫으려던 바로 그 창이 다시 열린다.
|
|
|
|
### 복합 키
|
|
|
|
`PRIMARY KEY (message_id, consumer_id)`.
|
|
|
|
message_id만으로 중복 제거하면 같은 event를 소비하는 두 번째 consumer가
|
|
첫 번째에 의해 억제된다. 각 consumer가 한 번씩 처리해야 한다.
|
|
|
|
### retention
|
|
|
|
broker의 최대 redelivery window보다 **길어야** 한다.
|
|
row를 먼저 지우면 늦게 도착한 redelivery가 두 번 처리된다.
|
|
|
|
## Debezium CDC 대안
|
|
|
|
polling relay 대신 WAL을 읽는다. polling interval과 lease 경합이 사라지지만
|
|
인프라와 그 자체의 실패 모드가 추가된다.
|
|
|
|
wire contract는 동일하다. `DebeziumOutboxEventRouter`가 polling relay와 같은 reserved header를
|
|
방출하므로 consumer는 어느 쪽이 발행했는지 구분할 수 없고, 전환은 배포 결정일 뿐 계약 변경이 아니다.
|
|
|
|
## Claim Check
|
|
|
|
1 MiB 초과 payload는 broker 프레임을 키우지 않고 외부 저장소로 offload한다.
|
|
|
|
`ClaimCheckReference`는 digest를 **필수**로 가진다. claim check는 메시지를 서로 다른 retention과
|
|
replication을 가진 두 시스템으로 쪼개므로, consumer는 producer가 저장한 바로 그 bytes를 받았음을
|
|
증명할 수 있어야 한다. 그렇지 않으면 잘린 객체와 정상 객체를 구분할 수 없다.
|
|
|
|
`ClaimCheckIntegrityGuard`는 fetch 전에 만료를, fetch 후에 크기와 digest를 검사한다.
|
|
digest 불일치는 `DESERIALIZATION`이 아니라 **validation** 실패로 분류한다.
|
|
bytes가 깨진 JSON인 게 아니라, 틀린 bytes이기 때문이다.
|