던플 개발기 (1): 시세 수집 파이프라인 — tier, 증분 저장, 멱등 워커

지난 편에서 사용자 조회 경로와 외부 수집 경로를 왜 갈라놨는지를 정리했습니다. 이번 편은 그 갈라진 한쪽, 수집 워커(market-cron)가 실제로 무엇을 하는가입니다.

수집은 단순히 "주기적으로 API를 호출해 DB에 넣는 일"처럼 보이지만, 막상 손대 보면 세 가지 질문이 동시에 달려듭니다. 누구를 언제 수집할 것인가, 워커를 여러 개 돌려도 같은 아이템을 중복 처리하지 않으려면 어떻게 하나, 그리고 매 분 들어오는 데이터를 어떻게 하면 적게 저장하면서도 시간축 조회를 지원하나. 이 글은 그 세 질문에 대한 답을 순서대로 풀어 갑니다.

한눈에 보면

• 던플마켓은 부동산 실거래 이력처럼 시간축으로 시세를 조회할 수 있어야 하지만, 네오플 API는 최근 100건·1개월·한 번에 400개 같은 한계가 있어 수집 워커가 필요합니다.
• 모든 아이템을 1분마다 수집하지 않습니다. 인기도에 따라 **tier(1·3·5·10분, DORMANT)**로 나눠 수집 간격을 정합니다.
• 검색 노출만으로는 수집하지 않고, 상세 페이지에 진입한 아이템부터 활성 수집 대상이 됩니다.
• 워커는 next_collect_at이 지난 아이템을 **lease 방식(FOR UPDATE SKIP LOCKED)**으로 집어, 여러 워커가 같은 아이템을 중복 처리하지 않습니다.
• SQS는 at-least-once라 모든 처리를 **멱등(idempotent)**하게 만들었습니다. 매 polling 전체를 복제하지 않고 새 매물·변경분만 증분 저장합니다.
• 상세 원본은 30일, 집계는 영구로 분리 보관합니다. 점검·장애를 0원이나 매물 0개로 기록하지 않습니다.

무엇을 풀어야 했나

요구는 분명했습니다. 사용자가 아이템 하나를 열면 현재 호가뿐 아니라 시간에 따른 시세 추이를 보고 싶어 합니다. 문제는 데이터 공급자인 네오플 API가 이 요구를 직접 채워 주지 않는다는 점입니다.

• GET /df/auction-sold(판매 시세)는 최근 100건 또는 최대 1개월 범위만 줍니다. 장기 차트를 만들려면 우리가 꾸준히 모아 둬야 합니다.
• GET /df/auction(현재 매물)은 한 번에 최대 400개 row만 반환합니다. 매물이 그보다 많으면 전체를 확정할 수 없습니다.
• 호출 한도(API key 기준 초당 500·분당 30,000)와 약관 제약이 있어, 마음껏 긁어올 수 없습니다.

여기서 가장 안일한 설계는 "활성 아이템을 1분마다 전부 호출해 매물 400개를 통째로 저장"하는 방식입니다. 이러면 활성 아이템이 늘수록 호출도 저장도 폭발합니다. 그래서 수집은 누구를 / 언제 / 무엇을 저장할지를 각각 제어하는 파이프라인이 됐습니다.

누구를 언제 수집할까: tier 모델

모든 아이템을 같은 주기로 수집하는 건 낭비입니다. 거래가 활발한 인기 아이템은 자주, 관심이 식은 아이템은 드물게 보면 됩니다. 그래서 활성 아이템에 수집 tier를 부여합니다.

활성화 규칙에는 한 가지 함정을 피하는 장치를 뒀습니다. 검색 자동완성에 노출됐다는 이유만으로 수집 대상이 되지는 않습니다. 입력만으로 활성 아이템이 무제한 늘어나기 때문입니다. 대신 사용자가 상세 페이지에 진입한 아이템을 활성화하고, 데이터가 오래됐으면 tier와 무관하게 즉시 한 번 수집한 뒤 WARM_5M으로 시작합니다.

tier는 고정이 아니라 인기 점수로 3분마다 다시 계산합니다. 점수는 세 신호를 종합합니다.

detail_view_count   상세 조회수
sold_quantity       판매 수량
sold_total_gold     총 거래 금액

여기서 골드 금액을 그대로 더하면 단위가 큰 값이 점수를 독점합니다. 그래서 각 신호에 log1p와 percentile 정규화를 적용한 뒤 가중치로 합칩니다.

normalized_detail_views  * 0.35
+ normalized_sold_quantity * 0.40
+ normalized_sold_total_gold * 0.25

window는 최근 24시간과 7일을 함께 봅니다. 점수가 낮고 최근 요청이 없는 아이템은 TTL 이후 DORMANT로 내려 주기 수집을 멈추고, 상세 조회나 찜으로 다시 깨어납니다. tier별 최대 아이템 수와 전체 호출 예산은 코드에 숫자를 흩뿌리지 않고 환경 설정으로 둡니다. 호출 한도 전부를 쓰지 않고, 워커에는 그보다 낮은 token bucket을 둬서 예산을 넘으면 무리하게 호출하는 대신 backlog로 남깁니다.

어떻게 중복 없이 돌릴까: dispatcher와 lease

아이템마다 EventBridge 스케줄을 만들지는 않습니다. 그러면 아이템이 늘 때마다 스케줄이 폭발합니다. 대신 1분에 한 번 전체 tick을 받고, 그 시점에 수집할 때가 된 아이템만 골라냅니다.

sequenceDiagram
  participant EB as EventBridge(1분)
  participant SQ as SQS scheduler
  participant DP as dispatcher
  participant PG as PostgreSQL
  participant CQ as SQS collect-item
  participant CO as collector
  participant NE as 네오플 API
  EB->>SQ: tick
  SQ->>DP: tick 수신
  DP->>PG: next_collect_at <= now() 아이템 claim (FOR UPDATE SKIP LOCKED)
  PG-->>DP: due 아이템 + lease
  DP->>CQ: collect-item job 발행
  CQ->>CO: job 수신
  CO->>NE: 매물·판매 이력 호출
  NE-->>CO: 응답
  CO->>PG: 트랜잭션으로 증분 저장 + next_collect_at 갱신

핵심은 dispatcher가 next_collect_at <= now()인 아이템을 FOR UPDATE SKIP LOCKED로 집는다는 점입니다. 이러면 워커(또는 dispatcher 인스턴스)를 늘려도 한 아이템을 한 워커만 집고, 나머지는 잠긴 row를 건너뜁니다. 집은 아이템에는 lease 만료 시각을 걸어, 워커가 중간에 죽어도 lease가 끝나면 다른 워커가 다시 가져갈 수 있습니다.

EventBridge는 60초 정밀도로 동작하므로 HOT_1M도 "정확히 매 정각 1분"이 아니라 "1분 주기를 목표로"입니다. 그래서 UI에는 항상 마지막 수집 시각을 함께 보여 줍니다. 거짓 정밀도를 만들지 않는 게 이 프로젝트의 일관된 원칙입니다.

어떻게 적게 저장할까: 증분 저장

SQS는 at-least-once 전달이라 같은 job이 두 번 올 수 있습니다. 따라서 수집 처리는 몇 번 실행해도 결과가 같아야(idempotent) 합니다. 그리고 매 polling마다 매물 400개를 통째로 다시 넣으면 저장량이 금세 터지므로, 현재 상태와 변경 이력을 분리해 새 매물·변경분만 추가합니다.

매 polling은 대략 이렇게 처리합니다. 매물마다 정규화 필드로 만든 listing_hash를 비교해, 새 매물이면 추가하고 바뀌었으면 갱신하고 그대로면 관측 시각만 손댑니다.

auction_snapshot_runs 생성
GET /df/auction 호출 후 정규화
각 매물에 대해:
  새 auction_no            → current insert + history DISCOVERED
  listing_hash 변경        → current update + history UPDATED
  변경 없음                → current.last_observed_at만 갱신
truncate 아닌 snapshot이면:
  사라진 매물 → ENDED 처리
minute aggregate upsert
snapshot_runs 완료

같은 매물이 30일 내내 노출돼도 변경이 없으면 history에는 최초 한 건(DISCOVERED)만 남습니다. 매 polling마다 400줄을 history에 다시 쌓지 않는 게 이 모델의 핵심입니다. seller처럼 제품에 필요 없는 값은 아예 저장하지 않고, listing_hash도 제품에 쓰는 정규화 필드로만 만듭니다.

여기서 400개 한계가 다시 등장합니다. 한 번에 400개만 받으니, 매물이 그보다 많으면 응답에 안 보이는 매물을 "팔렸다"고 단정할 수 없습니다.

• truncate되지 않은 성공 snapshot에서 사라진 매물만 종료(ENDED) 처리합니다.
• is_truncated = true인 snapshot에서는 누락 매물을 종료하지 않고 last_observed_at만 갱신합니다.
• 판매 이력과 이후의 비-truncate snapshot으로 종료 여부를 보완합니다.

종료 판정은 미묘해서 말로만 두지 않고 테스트로 고정했습니다. 판매 이력도 공급자가 주는 식별자와 판매 시각으로 중복을 막아, 같은 거래를 두 번 넣지 않습니다.

얼마나 오래 보관할까: 집계와 retention

원본을 영구 저장하면 데이터가 빠르게 커집니다. 그래서 상세 데이터는 짧게, 추세 데이터는 길게 나눠 둡니다.

집계 값에는 observed_listing_count, listing_average_unit_price, is_truncated처럼 "관측값"이라는 의미를 남기는 이름을 씁니다. API 한계로 전체 시장을 확정할 수 없으니, 평균이 아니라 "관측한 범위의 평균"임을 데이터 스스로 말하게 합니다.

flowchart LR
  RAW[원본 이력 + 분 집계<br/>30일, 일 단위 partition] --> ROLL[매일 rollup]
  ROLL --> AGG[시간·일·전체 집계<br/>영구]
  ROLL --> WM{rollup watermark<br/>검증}
  WM -->|성공| PURGE[30일 초과 partition 제거]
  WM -->|실패| KEEP[purge 중단·경고]

매일 도는 retention job은 순서가 중요합니다. 먼저 분 집계를 시간·일 집계로 rollup하고, rollup이 성공했는지(watermark)를 확인한 뒤에 30일을 넘긴 partition만 제거합니다. rollup이 실패하면 purge를 멈춥니다. 집계로 옮기지 못한 원본을 지워 버리면 데이터가 영영 사라지기 때문입니다. 최신 호가를 제공하는 auction_listings_current는 partition drop 대상이 아니라는 점도 분명히 해 뒀습니다.

정리하면

수집 파이프라인은 "API를 호출해 DB에 넣기"가 아니라, 외부 한계 안에서 시간축 데이터를 만들어 내는 일이었습니다.

• 누구를: 상세 진입으로 활성화하고, 인기 점수로 tier(1·3·5·10분/DORMANT)를 3분마다 재계산한다.
• 언제: 1분 tick에서 next_collect_at이 지난 아이템만 lease로 집어 중복 없이 처리한다.
• 무엇을: 매 polling 전체를 복제하지 않고 current/history를 분리해 증분 저장하며, 멱등하게 만든다.
• 얼마나: 상세 30일·집계 영구로 나누고, rollup 성공을 확인한 뒤에만 purge한다.

전반에 깔린 원칙은 하나입니다. 데이터가 거짓 정밀도를 갖지 않게 한다. 400개 한계는 is_truncated로, 수집 시점은 마지막 수집 시각으로, 관측 한계는 observed_* 이름으로 드러냅니다.

다음 편에서는 이 파이프라인이 외부 장애를 만났을 때 어떻게 버티는지 — 네오플 점검(503/DNF980)을 정상 운영 조건으로 다루는 법, circuit breaker, 그리고 점검 구간을 0으로 기록하지 않는 방법 — 을 다루겠습니다.

같이 보면 좋은 글

• 던플 개발기 (0): 모노레포·BFF·수집 분리를 이렇게 설계한 이유
• REST API는 HTTP의 의미를 어떻게 살린 설계일까