ファイルキューのコンシューマと冪等な再処理を実装する

목표

디렉터리 기반 큐로 소비자를 만들고, 파싱 실패를 격리하고, 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 에서 바로 읽는 것. 소비자를 두 개 띄우면 같은 메시지를 둘 다 처리합니다.

큐 디렉터리 구성

/root/q/inbox, /root/q/processing, /root/q/done, /root/q/error 를 만들고 /opt/lab/fixtures/eai/queue/inbox/ 의 모든 .json 을 /root/q/inbox 로 복사합니다. inbox 에 20개 파일이 있어야 합니다.

inbox/processing/done/error 네 구간을 만듭니다. 같은 파일시스템 안에 두어야 이동이 원자적이라는 점을 기억하세요.

메시지 구조 파악

메시지 구조를 파악해 /root/q/schema.csv 를 만듭니다. 첫 줄은 field,type,role. 정상 메시지에 있는 필드를 모두 적고, role 열에는 멱등키 역할을 하는 필드에 idempotency-key 를, 순서 판정에 쓰는 필드에 sequence 를 적습니다.

메시지 샘플을 열어 필드를 정리합니다. 어떤 필드가 멱등키가 될 수 있을지 눈여겨보세요.

소비자 구현과 실행

/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 은 비어 있어야 합니다.

파싱 실패는 재시도해도 똑같이 실패합니다. 그런 메시지는 즉시 error 로 보내야 뒤의 정상 메시지가 막히지 않습니다. 처리 이력은 업무 처리와 같은 트랜잭션에서 남기세요.

멱등성 확인

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

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

중복을 코드로 걸러 내는 것보다 DB 제약으로 막는 것이 안전합니다. 애플리케이션 버그가 있어도 제약은 뚫리지 않습니다.

실패 사유 분류

error 로 간 메시지의 사유를 /root/q/error.csv 에 정리합니다. 첫 줄은 file,reason. reason 은 parse 또는 missing-field 입니다.

사유를 '형식 오류'와 '업무 오류'로 나눠 보세요. 전자는 원천 수정이 필요하고 후자는 데이터 보완 후 재투입이 가능합니다.

순서 보장 검증

/root/q/order-check.sql 을 작성합니다. processed 테이블에서 같은 order_no 안에서 seq 가 중복되거나 비어 있는 경우를 찾는 쿼리입니다. 그 결과(문제 없으면 0행)를 /root/q/order-result.txt 에 저장합니다. 문제가 없으면 파일 첫 줄은 OK 여야 합니다.

같은 키의 메시지가 순번대로 처리됐는지를 SQL 로 확인합니다. 처리 시각이 아니라 저장된 순번을 비교해야 합니다.

적체 모니터링 스크립트

/root/q/lag.sh 를 만듭니다. 인자 두 개(큐디렉터리 임계치)를 받아 inbox=<n> processing=<n> error=<n> 한 줄을 출력하고, inbox 건수가 임계치를 초과하면 0 이 아닌 종료코드로 끝냅니다.

인자로 큐 디렉터리를 받아야 재사용됩니다. 임계치를 넘으면 종료코드로 알려야 크론이나 감시 도구에 붙일 수 있습니다.

재처리 스크립트

/root/q/replay.sh 를 만듭니다. 인자 하나(파일명)를 받아 error 의 그 파일을 inbox 로 되돌립니다. 해당 파일이 error 에 없으면 큐 상태를 전혀 바꾸지 말고 0 이 아닌 종료코드로 끝냅니다.

재처리는 아무거나 다시 넣는 것이 아닙니다. 존재하지 않는 메시지 ID 를 받으면 아무것도 하지 말고 실패해야 합니다.