[FEAT#524] Enrich stage Redis ZSET intermediate queue — priority sub-ordering

[juhy0987]

연관 이슈

• Closes [FEATURE] Enrich stage priority-aware 처리 — Redis ZSET intermediate queue #524
• Parent: [FEATURE] host/path 기반 priority 차등 배분 — publish 단계 + parser/validate/enrich 단계 priority-aware 처리 #515 (Phase 2 Sub 6, 마지막)
• 선행: [FEATURE] Parser stage priority-aware 처리 — Redis ZSET intermediate queue #522 (PR [FEAT#522] Parser stage Redis ZSET intermediate queue — priority sub-ordering #526), [FEATURE] Validate stage priority-aware 처리 — Redis ZSET intermediate queue #523 (PR [FEAT#523] Validate stage Redis ZSET intermediate queue — priority sub-ordering #527)

구현 내용

메타 이슈 #515 의 Phase 2 마지막 stage — Validate #527 패턴을 Enrich 에 미러링. Kafka partition FIFO 가 priority sub-ordering 을 제공 못 하는 한계를 Redis ZSET 으로 해소.

흐름 (Validate 동일):

Kafka.TopicValidated → [intake goroutine] → Redis ZSET → [worker pool] → process
                                                                  ↓ 실패
                                                           RetryScheduler.Enqueue → Kafka 재발행
                                                                  ↓ Enqueue 실패
                                                           DLQ fallback (메시지 영구 손실 방지)

Sub 1 — Enrich Worker retry hook (7830db9)

• Worker.retryScheduler 필드 + SetRetryScheduler setter
• Handle 분기: process 실패 → enqueueRetry → DLQ fallback → commit
• BuildRetryJob: ProcessingMessage → ContentRef → CrawlJob 변환
  • crawler/target_type 헤더 우선, "enrich-retry" fallback
  • PriorityFromHeader 헬퍼 / isValidTargetType 헬퍼
  • timeout_ms 헤더 계승
  • retry_reason="enrich_process_failed" + original_ref_id metadata

Sub 2 — Kafka → ZSET intake goroutine (1f4768f)

• worker.ZSetIntake — Validate 동일 패턴, 입력 토픽만 TopicValidated 로 변경
• 실패 정책 동일 (unmarshal/빈 ID → commit / push 실패 → commit skip)

Sub 3 — Feature flag + main.go wiring + Stage 통합 (e59474d)

• ENRICH_PRIORITY_QUEUE_ENABLED=true + STAGES_ENRICH_ENABLED=true + redisClientShared != nil 시 ZSET 모드 활성
• ZSET 모드 + retryScheduler nil → fatal
• enrich.Stage.SetZSetIntake 로 lifecycle 통합
• 환경변수: ENRICH_ZSET_QUEUE_KEY / ENRICH_ZSET_ENTRY_PREFIX / ENRICH_ZSET_MAX_SIZE / ENRICH_ZSET_ENTRY_TTL / ENRICH_ZSET_POP_TIMEOUT
• .env.example 갱신

Sub 4 — 단위 테스트 (이번 커밋)

• BuildRetryJob 11 cases + ZSetIntake.handleOne 6 cases + PriorityFromHeader 9 cases (총 26건)

CI / 머지 게이트 점검

변경 영향 범위

• 영향 패키지/모듈: internal/processor/enrich, cmd/issuetracker, .env.example
• 위험도: Medium — Enrich stage 핵심 흐름 변경. 단, feature flag default false + 미설정 시 기존 Kafka 흐름 100% 유지 → 안전 fallback.

Required Status Checks

• 통과 확인 대상 (PR Checks 탭에서 확인):
  • Commit Lint
  • PR Title Lint
  • Linked Issue Check
  • Format Check
  • Build
  • Test
  • Lint

롤백 계획

1. ENRICH_PRIORITY_QUEUE_ENABLED=false (또는 unset) — 기존 Kafka consume 흐름 복귀
2. ZSET 잔존 항목은 entry TTL 24h 후 자연 정리
3. 영구 롤백: 위 환경변수 + 본 PR revert

메타 이슈 #515 Phase 2 완료

본 PR 머지 시 메타 이슈 #515 의 모든 sub-issue (#521 / #522 / #523 / #524) 완료. Publisher 단계 host/path priority 분기 (Phase 1) + Parser / Validate / Enrich 의 ZSET priority sub-ordering (Phase 2) 인프라가 모두 갖춰집니다.

🤖 Generated with Claude Code

Summary by CodeRabbit

Release Notes

• New Features

  • Added optional Redis-backed priority queue support for the Enrich processing stage, enabling priority-based message routing and retry scheduling.
  • Priority queue automatically falls back to standard Kafka flow when not configured or unavailable.

• Configuration

  • New environment variables added for Enrich stage priority queue setup, including enabling the feature, queue sizing, and timeout controls.

[coderabbitai]

Warning

Rate limit exceeded

@juhy0987 has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 47 minutes and 24 seconds before requesting another review.

You’ve run out of usage credits. Purchase more in the billing tab.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 66be6de1-67c2-4b04-bcd6-e81a7def59e6

📥 Commits

Reviewing files that changed from the base of the PR and between 77c857d and 8481137.

📒 Files selected for processing (11)

• internal/processor/enrich/worker/worker.go
• internal/processor/enrich/worker/zset_intake.go
• internal/processor/parser/worker/worker.go
• internal/processor/parser/worker/zset_intake.go
• internal/processor/validate/worker/worker.go
• internal/processor/validate/worker/zset_intake.go
• pkg/queue/priority_header.go
• test/internal/processor/enrich/worker/retry_job_test.go
• test/internal/processor/parser/worker/retry_job_test.go
• test/internal/processor/validate/worker/retry_job_test.go
• test/pkg/queue/priority_header_test.go

📝 Walkthrough

Walkthrough

This PR extends the Enrich stage with a Redis ZSET intermediate queue for priority-aware message processing. When enabled, validated Kafka messages are immediately routed to a ZSET, consumed by workers with priority ordering, and failed messages are retried via a configurable scheduler instead of relying on Kafka redelivery.

Changes

Enrich Stage ZSET Priority Queue

Layer / File(s)	Summary
ZSET Configuration .env.example	Environment variables for ZSET queue operationalization: enable flag, Redis key/prefix, size limits, TTL, and pop timeout with fallback semantics.
Worker Retry Scheduling internal/processor/enrich/worker/worker.go, test/internal/processor/enrich/worker/retry_job_test.go	Worker accepts optional RetryScheduler injection; Handle branches on process failures to either enqueue retry jobs (reconstructed from message + headers with priority/crawler/type/timeout) or log-only; includes header parsers and target-type validation; comprehensive tests for job reconstruction across priority/crawler/timeout/type variations and edge cases.
ZSET Intake Worker internal/processor/enrich/worker/zset_intake.go, test/internal/processor/enrich/worker/zset_intake_test.go	New ZSetIntake component reads validated Kafka messages, unmarshals to extract content ID and priority, and pushes to priority queue with selective commit strategy: commits on unmarshal/validation failures (prevents redelivery loops), skips commit on push failures (allows Kafka redeliver); test suite validates all commit/push paths including success, unmarshal failures, empty IDs, push errors, and default priority handling.
Stage Integration & Lifecycle internal/processor/enrich/enrich.go, cmd/issuetracker/main.go	Stage gains optional ZSET intake injection via SetZSetIntake; Start launches intake concurrently. Main wiring conditionally creates ZSET consumer/queue/intake when feature enabled and Redis available; requires non-nil retry scheduler (Fatals otherwise) to prevent message loss during ZSET consumption; injects scheduler into worker and attaches intake to stage lifecycle.

Sequence Diagram(s)

sequenceDiagram
  participant KafkaConsumer as Kafka<br/>Consumer
  participant ZSetIntake
  participant PriorityZSetQueue
  participant Worker
  participant RetryScheduler
  participant Enricher
  
  KafkaConsumer->>ZSetIntake: FetchMessage (validated)
  ZSetIntake->>PriorityZSetQueue: Push(contentID, priority)
  PriorityZSetQueue-->>ZSetIntake: Enqueued
  ZSetIntake->>KafkaConsumer: Commit offset
  
  Worker->>PriorityZSetQueue: BZPOPMIN (priority-ordered)
  PriorityZSetQueue-->>Worker: Message with highest priority
  Worker->>Enricher: Process (LLM enrichment)
  
  alt Enrichment succeeds
    Enricher-->>Worker: Enriched content
    Worker->>Worker: Commit/store result
  else Enrichment fails
    Enricher-->>Worker: error
    Worker->>RetryScheduler: enqueueRetry(CrawlJob)
    RetryScheduler-->>Worker: Scheduled for retry
  end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

The PR spans multiple files with dense logic in worker retry handling and ZSET intake flow control. Worker error branching requires careful review of commit semantics, header parsing, and job reconstruction. ZSET intake implements a subtle selective-commit strategy across multiple failure modes. Wiring in main.go introduces new conditionals and fatality checks. Comprehensive test coverage mitigates some complexity. The changes are heterogeneous (worker behavior, new component, stage integration, configuration) rather than repetitive refactoring.

Possibly related PRs

• EinSofINTEREST/IssueTracker#527: Implements the same Redis ZSET intake pattern and RetryScheduler-driven retry logic for the Validate stage, using identical abstractions and error/commit policies.

Suggested labels

enhancement

🐰 A rabbit hops through Redis today,
Enrich stage queues by priority's way,
High scores pop first, retries rebound,
No Kafka loops when failures are found! ✨

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 19.35% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title '[FEAT#524] Enrich stage Redis ZSET intermediate queue — priority sub-ordering' directly and clearly describes the main change: implementing a Redis ZSET-based priority intermediate queue for the Enrich stage.
Linked Issues check	✅ Passed	All primary coding objectives from issue #524 are met: ZSET intake implementation (ZSetIntake reads Kafka, pushes to ZSET), worker retry hook (retryScheduler injection, BuildRetryJob conversion), ZSET feature flagging (ENRICH_PRIORITY_QUEUE_ENABLED), required env vars added, comprehensive unit tests (26 cases), and graceful shutdown via Stage.SetZSetIntake lifecycle integration.
Out of Scope Changes check	✅ Passed	All changes are in scope: .env.example additions (ZSET env vars), cmd/issuetracker/main.go (ZSET mode wiring), internal/processor/enrich/* (Stage, Worker, ZSetIntake, tests). No out-of-scope modifications to unrelated subsystems detected.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feature/#524/enrich-priority-zset-queue

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

internal/processor/enrich/enrich.go (1)

45-55: ⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

Stop does not own intake shutdown lifecycle

Start launches s.intake.Run(ctx) but Stop only stops s.worker. If Stop is called before the start-context is canceled, intake can keep consuming in the background.

Suggested lifecycle fix

 type Stage struct {
   worker *worker.Worker
   intake *worker.ZSetIntake // nil 허용 — ZSET 인입 모드일 때만 (이슈 `#524`)
+  intakeCancel context.CancelFunc
+  intakeDone   chan struct{}
 }

 func (s *Stage) Start(ctx context.Context) {
   s.worker.Start(ctx)
   if s.intake != nil {
-    go s.intake.Run(ctx)
+    intakeCtx, cancel := context.WithCancel(ctx)
+    s.intakeCancel = cancel
+    s.intakeDone = make(chan struct{})
+    go func() {
+      defer close(s.intakeDone)
+      s.intake.Run(intakeCtx)
+    }()
   }
 }

 func (s *Stage) Stop(ctx context.Context) error {
+  if s.intakeCancel != nil {
+    s.intakeCancel()
+    if s.intakeDone != nil {
+      select {
+      case <-s.intakeDone:
+      case <-ctx.Done():
+        return ctx.Err()
+      }
+    }
+  }
   return s.worker.Stop(ctx)
 }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@internal/processor/enrich/enrich.go` around lines 45 - 55, The Stop method
currently only stops s.worker but doesn't stop the intake goroutine started in
Start (s.intake.Run(ctx)), so update Stop (on type Stage) to also shut down the
intake when present: if s.intake != nil call the intake's shutdown method (e.g.,
s.intake.Stop(ctx) or signal/cancel the intake run) and wait for it to complete
before/alongside s.worker.Stop(ctx); ensure you reference the same symbols
(Stage.Stop, Stage.Start, s.intake.Run, s.worker.Stop) and choose the intake's
existing stop/close API or add one if missing to properly own the intake
lifecycle.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@internal/processor/enrich/enrich.go`:
- Around line 45-55: The Stop method currently only stops s.worker but doesn't
stop the intake goroutine started in Start (s.intake.Run(ctx)), so update Stop
(on type Stage) to also shut down the intake when present: if s.intake != nil
call the intake's shutdown method (e.g., s.intake.Stop(ctx) or signal/cancel the
intake run) and wait for it to complete before/alongside s.worker.Stop(ctx);
ensure you reference the same symbols (Stage.Stop, Stage.Start, s.intake.Run,
s.worker.Stop) and choose the intake's existing stop/close API or add one if
missing to properly own the intake lifecycle.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 2115e426-f04c-4416-bc3e-6645aa8208b9

📥 Commits

Reviewing files that changed from the base of the PR and between 83be71c and 77c857d.

📒 Files selected for processing (7)

• .env.example
• cmd/issuetracker/main.go
• internal/processor/enrich/enrich.go
• internal/processor/enrich/worker/worker.go
• internal/processor/enrich/worker/zset_intake.go
• test/internal/processor/enrich/worker/retry_job_test.go
• test/internal/processor/enrich/worker/zset_intake_test.go

[gemini-code-assist]

Code Review

This pull request introduces a Redis ZSET-based priority queue for the Enrich stage, enabling priority sub-ordering by bypassing Kafka's partition FIFO constraints. Key changes include the implementation of a ZSetIntake component to move messages from Kafka to Redis, the addition of a RetryScheduler to the Enrich worker to prevent message loss in ZSET mode, and the necessary wiring in the main entry point. Feedback highlights the need to ensure at-least-once processing by avoiding offset commits if retry or DLQ operations fail, the recommendation to use context.WithoutCancel for terminal operations, and the opportunity to refactor the duplicated PriorityFromHeader logic into a common utility package.