[FEATURE] host/path 기반 priority 차등 배분 — publish 단계 + parser/validate/enrich 단계 priority-aware 처리

[juhy0987]

배경

이슈 #503 ~ #507 작업 흐름에서 확인된 현 운영의 priority 분포:

토픽	트래픽
crawl.high	0건 (시드 0건 + chained/upgrade 모두 명시 안 함)
crawl.normal	95%+ — Seed Normal 시드 + Chained 모든 article + Upgrade 하드코딩
crawl.low	거의 0 (google_cse 1건만, interval 6h)

원인 (publisher level):

• bus.SourcePriorityResolver / RuleBasedPriorityResolver 가 wiring 만 됐고 룰 등록 0건 (dead wiring)
• chain.go:buildJobMessages 가 job.Priority 미설정 → DefaultPriorityResolver → Normal
• upgrader.go:223 가 priority 하드코딩 → Normal

원인 (stage level):

• parser / validate / enrich 가 단일 입력 토픽 (TopicFetched / TopicNormalized / TopicValidated) 소비
• 단일 토픽 = Kafka partition 순서 FIFO → 같은 stage 안에서 priority sub-ordering 부재
• Kafka header priority 는 보존되지만 라우팅·정렬에 미사용

→ 결과: priority 분기 메커니즘이 이론적으로 존재 (ExplicitPriorityResolver + 3 토픽 × 3 pool 격리) 이지만 실제 운영에서 차등 적용 안 됨.

변경 방향

Two-track 접근:

1. Publisher 단계 — 다음 URL 을 publish 할 때 host/path 기반으로 priority 자동 결정
2. Stage 단계 — parser/validate/enrich 가 priority 를 인식해 차등 처리

Track 1 — Publisher 단계 (host/path → priority 매핑)

1-A. Priority 룰 저장 — 옵션 결정 필요

옵션 A (권장): 기존 parser_rules 에 컬럼 추가

ALTER TABLE parser_rules ADD COLUMN crawl_priority SMALLINT NOT NULL DEFAULT 2;
-- 1=high / 2=normal / 3=low — scheduler_entries.priority 와 동일 매핑

• 장점: host_pattern + path_pattern 이 이미 있음 — RuleBasedPriorityResolver 와 자연 1:1
• 단점: parser_rules 의 의미 확장 (selector 외 라우팅 정책 포함)

옵션 B: 신규 crawl_priority_rules 테이블

CREATE TABLE crawl_priority_rules (
  id SERIAL PRIMARY KEY,
  host_pattern TEXT NOT NULL,
  path_pattern TEXT NOT NULL DEFAULT '',
  priority SMALLINT NOT NULL CHECK (priority BETWEEN 1 AND 3),
  enabled BOOLEAN NOT NULL DEFAULT TRUE,
  notes TEXT,
  ...
);

• 장점: 책임 분리 (parser_rules 는 selector 만, 본 테이블은 라우팅만)
• 단점: 룰 lookup 이중화

옵션 C: 정적 config (fetcher_rules.yaml) 또는 env

• 단점: 운영 중 변경 어려움 — 본 이슈의 목적 (동적 차등) 약화

1-B. RuleBasedPriorityResolver 룰 등록

internal/bus/resolver.go:140+ — AddRule(rule) 메소드 사용해 DB lookup 결과를 wiring 시점에 또는 runtime cache 로 등록.

ruleResolver := bus.NewRuleBasedPriorityResolver(core.PriorityNormal)
for _, r := range priorityRules {
    ruleResolver.AddRule(bus.PriorityRule{
        Match: func(job *core.CrawlJob) bool {
            // host_pattern + path_pattern regex 매칭
            return matchHostPath(r.HostPattern, r.PathPattern, job.Target.URL)
        },
        Priority: priorityFromInt(r.Priority),
    })
}
resolver.Add(ruleResolver)

1-C. Chained job 의 priority 명시

internal/bus/chain.go:96+ — buildJobMessages 가 현재 job.Priority 미설정. 옵션:

• (a) 매 URL 마다 resolver chain 통과 (현재 buildMessage 가 이미 함) — 별도 변경 불필요. RuleBased 룰 등록만 하면 자동 작동
• 즉 본 sub-task 는 1-B 가 wiring 되면 자연 해결

1-D. Upgrade 의 priority 하드코딩 제거

internal/processor/fetcher/rule/upgrader.go:223,235 — Priority: core.PriorityNormal + Topic: queue.TopicCrawlNormal 하드코딩.

• 변경: buildMessage 경유하도록 PublishUpgrade 리팩터 → resolver chain 통과
• 또는 원 raw_id 의 host 기반 lookup

Track 2 — Stage 단계 priority-aware 처리

parser / validate / enrich 가 현재 단일 토픽 소비. priority 차등 처리 옵션:

옵션 2-A (권장): 토픽 3분 — fetcher 패턴 그대로

각 stage 도 <stage>.high / <stage>.normal / <stage>.low 3개 토픽 + 3개 worker pool 분리.

Stage	현재	신규
Parser	issuetracker.fetched × 1	issuetracker.fetched.{high,normal,low} × 3
Validate	issuetracker.normalized × 1	issuetracker.normalized.{high,normal,low} × 3
Enrich	issuetracker.validated × 1	issuetracker.validated.{high,normal,low} × 3

• 장점: fetcher 와 동일 패턴, resource isolation 보장 (High 항상 즉시 처리)
• 단점:
  • 토픽 수 증가 (4 → 12 토픽)
  • 환경변수 *_HIGH_WORKER_COUNT / NORMAL / LOW 신설
  • 메시지 발행 시점에 priority 결정 (RawContentRef / ContentRef Kafka 헤더 사용)
  • migration cost — 기존 consumer/producer wiring 전반 수정

옵션 2-B: In-memory priority queue per worker

Worker pool 이 단일 토픽 consume 후 in-process priority queue (heap) 에 적재, worker 가 priority 순으로 pop.

• 장점: 토픽 변경 없음 — 기존 인프라 유지
• 단점:
  • 인스턴스별 격리 (다른 인스턴스의 High 를 못 봄)
  • graceful shutdown 시 in-memory 잔존물 손실
  • 단일 인스턴스에서만 효과적

옵션 2-C: Redis ZSET intermediate queue

Stage 별로 Kafka → Redis ZSET (score=priority+timestamp) → worker pop.

• 이슈 [FEATURE] normal/low crawl 토픽 Redis 버퍼링 + 주기적 Kafka backlog drain #510 (PR [FEAT#510] normal/low crawl 토픽 Redis 버퍼링 + 주기적 Kafka backlog drain #511) 의 BufferingProducer + BufferDrainer 와 동일 패턴 재사용 가능
• 장점: 다중 인스턴스에서도 priority 일관, 기존 토픽 그대로
• 단점: Redis 부하 + complexity

권장 phasing

본 이슈는 substantial — sub-issue 로 분리 권장:

1. Sub 1 [FEAT]: Priority 룰 저장 스키마 (옵션 A or B 결정)
2. Sub 2 [FEAT]: RuleBasedPriorityResolver DB-backed 룰 lookup + cache + wiring
3. Sub 3 [FIX]: Upgrade 하드코딩 priority 제거 + buildMessage 경유
4. Sub 4 [FEAT]: Parser stage priority-aware (옵션 2-A or 2-C 결정)
5. Sub 5 [FEAT]: Validate stage priority-aware
6. Sub 6 [FEAT]: Enrich stage priority-aware

Sub 13 만 먼저 진행하면 Publisher 단계의 priority 차등은 즉시 효과 — Stage 단계 (Sub 46) 는 별도 phase.

완료 조건

• Sub 1: priority 룰 저장 스키마 + migration (옵션 결정)
• Sub 2: RuleBasedPriorityResolver DB lookup + main.go wiring
  • 라이브에서 chained article URL 이 host/path 기반 priority 로 분기되는지 확인
• Sub 3: PublishUpgrade 하드코딩 제거 — buildMessage 경유
• Sub 4~6: parser/validate/enrich stage priority-aware 처리 (옵션 결정 후 phasing)
• 단위 테스트: priority 룰 매칭 (host regex / path regex 우선순위) + resolver chain order
• 통합 테스트: 라이브 동등 시나리오에서 priority 분기 검증
• .env.example + 운영 문서 갱신 (priority 룰 설정 가이드)

Why

• 현 운영에서 95%+ 트래픽이 Normal 토픽 단일 경로 → priority 분기 인프라가 사실상 미사용
• breaking-news / urgent-update / archive 등 호스트/path 별 처리 우선순위 차등이 시스템 자원 활용 효율 향상
• ExplicitPriorityResolver 만 동작 중 — 다른 두 resolver 는 dead wiring, 본 이슈가 활성화

우선순위

Medium — 현재 시스템이 priority 분기 없이도 정상 동작 중 (resource isolation 으로 High 가 굶지 않음). 그러나 미래 확장 (breaking-news 소스 추가 등) 시 본 차등이 핵심.

관련

• 라이브 회차: 2026-05-19T00:25:00 (PR [REFAC#503] rate_limit 로그에 label 필드 추가 — throttled 호스트/IP 사후 추적 가능 #514 머지 직후)
• 이슈 [FEATURE] normal/low crawl 토픽 Redis 버퍼링 + 주기적 Kafka backlog drain #510 (PR [FEAT#510] normal/low crawl 토픽 Redis 버퍼링 + 주기적 Kafka backlog drain #511) — BufferingProducer + BufferDrainer (옵션 2-C 의 토대)
• 이슈 [FEATURE] cmd/issuetracker stage-toggle env flags — 단일 바이너리로 stage 별 독립 실행 지원 #443 — stage toggle env (다중 인스턴스 운영 시발점)
• bus.PriorityResolver chain — internal/bus/resolver.go
• chained job priority 미설정: internal/bus/chain.go:96+
• upgrade priority 하드코딩: internal/processor/fetcher/rule/upgrader.go:223,235
• parser_rules host_pattern + path_pattern 컬럼 (옵션 A 활용 후보)

[juhy0987]

메타 이슈 #515 완료 — Phase 1 + Phase 2 전 sub-issue 머지

전체 4개 sub-issue + 4개 PR 머지로 host/path 기반 priority 차등 배분 인프라가 publisher → parser → validate → enrich 파이프라인 전반에 적용 완료.

Phase 1 — Publisher 단계 (단일 PR)

#521 → PR #525

• migration 032: parser_rules.crawl_priority SMALLINT NOT NULL DEFAULT 2 (1=high/2=normal/3=low) + CHECK 제약
• RuleBasedPriorityResolver DB-backed hydrate + 주기 refresh (env PRIORITY_RULES_REFRESH_INTERVAL, default 5분)
• bus/priority_refresher.go 신설 — atomic.Pointer 로 lock-free 룰 교체
• upgrader.go priority/topic 하드코딩 제거 → bus.PriorityResolver 경유
• 핵심 효과: chained article URL + upgrade URL 이 host/path 기반으로 crawl.high/normal/low 토픽 자동 라우팅

Phase 2 — Stage 단계 (3 PR, Redis ZSET intermediate queue)

공통 인프라 (#522 PR #526 도입)

• pkg/queue.PriorityZSetQueue — score = priority × 1e13 + arrival_ts_ms
• pkg/queue.PriorityZSetConsumer (queue.Consumer 어댑터)
• pkg/queue.PriorityPusher 인터페이스 (push-only, 단위 테스트 용이)
• pkg/queue.PriorityFromHeader 단일 출처 (DRY 적용)
• BZPOPMIN atomic pop + GET 실패 시 ZADD 복구 (메시지 손실 방지)
• MaxSize 초과 시 ZREMRANGEBYRANK (low priority 항목 drop)

#522 Parser → PR #526

• Parser ZSet 인입 모드 (TopicFetched → Redis ZSET)
• Worker.SetRetryScheduler + Handle retry hook
• BuildRetryJob: RawContentRef → CrawlJob, headers 우선 (crawler / target_type / priority)
• Kafka → ZSET intake goroutine
• PARSER_PRIORITY_QUEUE_ENABLED feature flag

#523 Validate → PR #527

• Validate ZSet 인입 모드 (TopicNormalized → Redis ZSET)
• ProcessingMessage + ContentRef payload 대응
• DLQ fallback (enqueueRetry 실패 시 메시지 영구 손실 방지)
• timeout_ms 헤더 계승 (buildRetryDefaultTimeout 30s fallback)
• STAGES_VALIDATE_ENABLED 가드
• VALIDATE_PRIORITY_QUEUE_ENABLED feature flag

#524 Enrich → PR #528

• Enrich ZSet 인입 모드 (TopicValidated → Redis ZSET)
• Validate 패턴 그대로 미러링 (ContentRef payload + DLQ fallback)
• PriorityFromHeader 의 pkg/queue 단일 출처 이관 (3개 worker 중복 제거)
• ENRICH_PRIORITY_QUEUE_ENABLED feature flag

핵심 정량 지표

항목	값
sub-issue 수	4 (#521 / #522 / #523 / #524)
머지 PR 수	4 (#525 / #526 / #527 / #528)
신규 코드	~2,000 LoC + 단위 테스트 ~1,200 LoC
신규 환경변수	18개 (Parser/Validate/Enrich × 6 + PRIORITY_RULES_REFRESH_INTERVAL/LOAD_LIMIT)
migration	032 (parser_rules.crawl_priority)
신규 Kafka 토픽	0 (ZSET intermediate queue 로 토픽 확장 회피)
신규 Redis ZSET	3 (parser:zset:queue / validate:zset:queue / enrich:zset:queue)

후속 작업

• #529 (Bug) — Parser/Validate/Enrich Stage.Stop 이 ZSetIntake goroutine 종료를 명시적으로 wait 하지 않음. graceful shutdown 일관성 강화 필요.

정책 결정 사항 (작업 도중)

• schema 옵션: parser_rules 컬럼 추가 (옵션 A) — 신규 테이블보다 host_pattern + path_pattern + crawl_priority 1:1 자연
• stage 메커니즘: Redis ZSET (옵션 2-C) — Kafka 토픽 분리 (옵션 2-A) 대비 토픽 인프라 변경 0 + 다중 인스턴스 priority 일관
• PriorityFromHeader 위치: pkg/queue (DRY — gemini PR [FEAT#524] Enrich stage Redis ZSET intermediate queue — priority sub-ordering #528 피드백)
• 메시지 손실 방지: enqueueRetry 실패 시 DLQ fallback (coderabbit PR [FEAT#523] Validate stage Redis ZSET intermediate queue — priority sub-ordering #527 #3275227536)
• priorityFactor: 1e10 → 1e13 (gemini PR [FEAT#522] Parser stage Redis ZSET intermediate queue — priority sub-ordering #526 high-priority — UnixMilli 대비 priority dominant 보장)

활성화 조건 (각 stage 공통):

1. *_PRIORITY_QUEUE_ENABLED=true
2. STAGES_*_ENABLED=true
3. Redis 가용 (redisClientShared != nil)
4. RetryScheduler 설정 (Redis 가용 시 자동 wiring)

미충족 시 안전 fallback (기존 Kafka 직접 consume 흐름) — feature flag default false 로 운영 호환 보장.