파일 큐 소비자와 멱등 재처리 구현

목표

디렉터리 기반 큐로 소비자를 만들고, 파싱 실패를 격리하고,
DB 제약으로 멱등성을 보장하고, 적체 감시와 재처리까지 구현할 수 있게 됩니다.

왜 중요한가

비동기 연동에서 중복은 예외가 아니라 기본값입니다. 'exactly-once' 는
전달 계층의 마법이 아니라 '적어도 한 번 전달 + 수신측 중복 제거'의 결과입니다.
그래서 소비자는 언제나 중복을 가정하고 짜야 합니다.
그리고 파싱 실패 메시지를 재시도 큐에 넣으면 그 메시지가 맨 앞에서 계속 실패하며
뒤의 정상 메시지를 막습니다(독약 메시지). 막힌 큐는 곧 업무 정지입니다.
이 두 가지를 손으로 구현해 보면 비동기 설계의 감이 생깁니다.

단계

1. /root/q/inbox, /root/q/processing, /root/q/done, /root/q/error 를 만들고
/opt/lab/fixtures/eai/queue/inbox/ 의 모든 .json 을 /root/q/inbox 로 복사합니다.
inbox 에 20개 파일이 있어야 합니다.
2. 메시지 구조를 파악해 /root/q/schema.csv 를 만듭니다.
첫 줄은 field,type,role. 정상 메시지에 있는 필드를 모두 적고,
role 열에는 멱등키 역할을 하는 필드에 idempotency-key 를,
순서 판정에 쓰는 필드에 sequence 를 적습니다.
3. /root/q/consume.py 를 만들고 실행합니다. 동작은 아래와 같습니다.

• inbox 의 파일을 하나씩 processing 으로 옮긴 뒤 읽습니다.
• JSON 파싱 실패 또는 필수 필드(msg_id, order_no, seq, amount) 누락이면

error 로 옮깁니다.

• 정상이면 sqlite DB /root/q/ledger.db 의 processed 테이블에 적재하고

done 으로 옮깁니다.

• processed 테이블은 msg_id 를 PRIMARY KEY 또는 UNIQUE 로 가져야 하고,

order_no, seq, amount, processed_at 컬럼이 있어야 합니다.

• 실행이 끝나면 processing 은 비어 있어야 합니다.

4. 실행 결과가 아래와 같아야 합니다.

• done 18개, error 2개
• processed 테이블 행 수는 중복 제거된 15건
• 중복으로 무시된 msg_id 3건을 /root/q/dup.txt 에 오름차순으로 저장

5. error 로 간 메시지의 사유를 /root/q/error.csv 에 정리합니다.
첫 줄은 file,reason. reason 은 parse 또는 missing-field 입니다.
6. /root/q/order-check.sql 을 작성합니다. processed 테이블에서
같은 order_no 안에서 seq 가 중복되거나 비어 있는 경우를 찾는 쿼리입니다.
그 결과(문제 없으면 0행)를 /root/q/order-result.txt 에 저장합니다.
문제가 없으면 파일 첫 줄은 OK 여야 합니다.
7. /root/q/lag.sh 를 만듭니다. 인자 두 개(큐디렉터리 임계치)를 받아
inbox=<n> processing=<n> error=<n> 한 줄을 출력하고,
inbox 건수가 임계치를 초과하면 0 이 아닌 종료코드로 끝냅니다.
8. /root/q/replay.sh 를 만듭니다. 인자 하나(파일명)를 받아
error 의 그 파일을 inbox 로 되돌립니다.
해당 파일이 error 에 없으면 큐 상태를 전혀 바꾸지 말고
0 이 아닌 종료코드로 끝냅니다.

참고

• sqlite 스키마 확인: sqlite3 /root/q/ledger.db '.schema processed'
• 중복 삽입 무시: INSERT OR IGNORE 또는 ON CONFLICT DO NOTHING
• 원자적 이동: 같은 파일시스템 안에서 os.rename / mv
• 흔한 실수 1: 파싱 실패 메시지를 다시 inbox 로 돌려보내 무한 루프를 만드는 것.
• 흔한 실수 2: 중복 방지를 애플리케이션 조건문으로만 하는 것.

동시 실행되면 뚫립니다. DB 제약이 마지막 방어선입니다.

• 흔한 실수 3: processing 을 안 쓰고 inbox 에서 바로 읽는 것.

소비자를 두 개 띄우면 같은 메시지를 둘 다 처리합니다.

단계 8개

1. 큐 디렉터리 구성
2. 메시지 구조 파악
3. 소비자 구현과 실행
4. 멱등성 확인
5. 실패 사유 분류
6. 순서 보장 검증
7. 적체 모니터링 스크립트
8. 재처리 스크립트