[FEAT#522] Parser stage Redis ZSET intermediate queue — priority sub-ordering

[juhy0987]

연관 이슈

• Closes [FEATURE] Parser stage priority-aware 처리 — Redis ZSET intermediate queue #522
• Parent: [FEATURE] host/path 기반 priority 차등 배분 — publish 단계 + parser/validate/enrich 단계 priority-aware 처리 #515 (Phase 2 Sub 4)

구현 내용

메타 이슈 #515 의 Phase 2 Parser stage. Kafka partition FIFO 가 priority sub-ordering 을 제공 못 하는 한계를 Redis ZSET intermediate queue 로 해소. PR #511 / #516 의 Redis 패턴 재사용.

흐름:

Kafka.TopicFetched → [intake goroutine] → Redis ZSET → [worker pool] → ProcessMessage
                                                              ↓ 실패
                                                       RetryScheduler.Enqueue → Kafka 재발행

Sub 1 — pkg/queue.PriorityZSetQueue 추상화 (fd91b3c)

• score = priority(1=high/2=normal/3=low) × 1e10 + arrival_timestamp_ms
• Push(ctx, priority, id, payload) — ZADD + SET pipeline (retry_queue 동일 패턴)
• Pop(ctx, timeout) — BZPOPMIN + entry GET / DEL (atomic pop = ack)
• Len(ctx) — ZCard
• MaxSize 초과 시 ZREMRANGEBYRANK 로 가장 낮은 priority + 오래된 항목 drop
• PriorityZSetConsumer — queue.Consumer 어댑터 → workerpool.ConsumerPool 재사용
• 16 단위 테스트 (Redis 통합, 미가용 시 skip)

Sub 2 — Parser Worker RetryScheduler hook (3e5e285)

• Worker.retryScheduler bus.RetryScheduler 필드 + SetRetryScheduler setter
• Handle 분기: ProcessMessage 실패 시
  • retryScheduler 주입 시 → enqueueRetry 후 commit (메시지 손실 방지)
  • 미주입 시 → commit skip → Kafka redeliver (기존 동작 호환)
• BuildRetryJob(msg) (*core.CrawlJob, error) 헬퍼 (export for test)

Sub 3 — Kafka → ZSET intake goroutine (b724732)

• worker.ZSetIntake 단일 goroutine
• FetchMessage → Unmarshal → priority 추출 → ZSET Push → Kafka commit
• 실패 정책: unmarshal/빈 ID → commit (재시도 무의미) / ZSET push 실패 → commit skip (Kafka redeliver) / commit 실패 → 다음 fetch (idempotent)
• PriorityFromHeader(headers) 헬퍼 (export for test)

Sub 4 — Feature flag + main.go wiring (370b0ff)

• PARSER_PRIORITY_QUEUE_ENABLED=true + redisClientShared != nil 시 ZSET 모드 활성
• ZSET 모드 + retryScheduler nil → fatal (메시지 손실 방지 가드)
• parser.Stage.SetZSetIntake 로 lifecycle 통합 (Stage.Start 시 go intake.Run(ctx), ctx cancel 시 자연 종료)
• 환경변수: PARSER_ZSET_QUEUE_KEY / PARSER_ZSET_ENTRY_PREFIX / PARSER_ZSET_MAX_SIZE / PARSER_ZSET_ENTRY_TTL / PARSER_ZSET_POP_TIMEOUT
• envBoolOrDefault / envOrDefault 헬퍼 추가

Sub 5 — 단위 테스트 (134b3db)

• BuildRetryJob 7 cases: 정상 / 잘못된 priority / 빈 source name / 빈 URL / malformed JSON
• PriorityFromHeader 9 cases: 1/2/3 매핑 / missing / empty / non-numeric / out-of-range
• pkg/queue PriorityZSet 16 cases (통합)

CI / 머지 게이트 점검

변경 영향 범위

• 영향 패키지/모듈: pkg/queue, internal/processor/parser/worker, internal/processor/parser, cmd/issuetracker
• 위험도: Medium — Parser 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. PARSER_PRIORITY_QUEUE_ENABLED=false (또는 unset) — 기존 Kafka 직접 consume 흐름 복귀
2. Redis 의 잔존 ZSET 항목은 다음 부팅 시 자동 처리 (entry TTL 24h 후 자연 정리)
3. 영구 롤백: 위 환경변수 + 본 PR revert

후속 작업 (메타 #515 Phase 2)

• [FEATURE] Validate stage priority-aware 처리 — Redis ZSET intermediate queue #523 Validate stage Redis ZSET (선행: 본 PR 의 pkg/queue.PriorityZSetQueue 추상화 활용)
• [FEATURE] Enrich stage priority-aware 처리 — Redis ZSET intermediate queue #524 Enrich stage Redis ZSET (선행: 본 PR, 병렬 가능)

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features
  • Optional priority-based message processing mode (configurable via environment settings)
  • Enhanced handling of transient failures with automatic retry capability to prevent message loss

[coderabbitai]

Warning

Rate limit exceeded

@juhy0987 has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 45 minutes and 59 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: d86fdfbb-1fd5-4338-a5b2-726080327b00

📥 Commits

Reviewing files that changed from the base of the PR and between 134b3db and ae1f02e.

📒 Files selected for processing (7)

• .env.example
• internal/processor/parser/worker/worker.go
• internal/processor/parser/worker/zset_intake.go
• pkg/queue/priority_zset.go
• test/internal/processor/parser/worker/retry_job_test.go
• test/internal/processor/parser/worker/zset_intake_test.go
• test/pkg/queue/priority_zset_test.go

📝 Walkthrough

Walkthrough

This PR implements an optional Redis ZSET-backed priority queue for the parser stage, allowing priority-aware message processing independent of Kafka partition ordering. When enabled via env flag, Kafka messages are bridged into a Redis ZSET, popped by priority + arrival order, with retry-scheduler integration for fail-safe redelivery without message loss.

Changes

Parser Priority Queue via Redis ZSET

Layer / File(s)	Summary
Redis ZSET Priority Queue Foundation pkg/queue/priority_zset.go, test/pkg/queue/priority_zset_test.go	PriorityZSetQueue computes ZSET scores from priority and timestamp, enforces max-size overflow by dropping high-score entries, and provides atomic pop via BZPopMin. PriorityZSetConsumer adapts the queue to the consumer interface with configurable poll timeout and header-based priority recovery. Complete test coverage for roundtrips, priority ordering, FIFO within priority, timeout/cancellation behavior, and edge cases.
Parser Worker Retry Support internal/processor/parser/worker/worker.go, test/internal/processor/parser/worker/retry_job_test.go	Worker gains optional retryScheduler field and SetRetryScheduler method. On transient Handle errors, when scheduler is present, BuildRetryJob converts the failed message to a CrawlJob with parsed priority header (defaulting to normal), preserved crawler name, and retry metadata, enqueued via scheduler before committing the original Kafka message. Comprehensive tests verify retry job construction, priority header parsing, fallback behavior, and validation.
Kafka→ZSET Intake Bridge internal/processor/parser/worker/zset_intake.go	ZSetIntake worker continuously fetches Kafka messages, unmarshals RawContentRef, derives priority via PriorityFromHeader, and pushes into the ZSET queue. Differentiates failure modes: JSON/empty-ID errors are committed immediately to avoid redelivery loops; ZSET push failures skip commit to allow Kafka retry; Kafka commit failures assume idempotency.
Parser Stage Lifecycle Integration internal/processor/parser/stage.go	Stage gains optional intake field and SetZSetIntake method to accept the ZSET intake component. Start conditionally launches the intake goroutine when present, tying its lifecycle to stage context cancellation without requiring a separate Stop.
Configuration, Feature Gating & Main Wiring cmd/issuetracker/main.go	ZSET mode is gated by PARSER_PRIORITY_QUEUE_ENABLED env var and Redis availability. When enabled, PriorityZSetQueue and ZSetIntake are constructed with env-configured keys, TTL, and size limits; parser worker consumes from ZSET consumer; retryScheduler is injected into worker with startup hard-fail if scheduler is unconfigured. Parser stage receives the intake via SetZSetIntake for lifecycle binding. New envBoolOrDefault and envOrDefault helpers support configuration.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related issues

• EinSofINTEREST/IssueTracker#515: This PR implements Option 2-C (Kafka→Redis ZSET intermediate queue) described in the parent issue for making the parser stage priority-aware.
• EinSofINTEREST/IssueTracker#522: This PR directly fulfills the feature request for Redis ZSET intermediate queue with parser priority-aware processing.
• EinSofINTEREST/IssueTracker#523: Related retrieve issue that reuses the PriorityZSetQueue and intake patterns introduced in this PR.

Possibly related PRs

• EinSofINTEREST/IssueTracker#400: Both PRs wire the global retryScheduler in main.go; this PR requires the same scheduler for ZSET-mode fail-safe redelivery.

Suggested labels

enhancement

Poem

🐰 A fluffy queue in Redis springs,
Priorities sorted by ZSET wings,
Kafka feeds the queue with care,
Workers pop with graceful flair.
Retry paths keep messages whole—
The ZSET takes priority's role! 🌟

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 34.21% 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#522] Parser stage Redis ZSET intermediate queue — priority sub-ordering' accurately reflects the main change: implementing a Redis ZSET-based priority queue for the parser stage to enable priority-aware ordering.
Linked Issues check	✅ Passed	The PR fully implements all coding-related requirements from issue #522: PriorityZSetQueue implementation, Kafka→ZSET intake with proper commit handling, worker RetryScheduler injection, BuildRetryJob conversion, and comprehensive unit tests for priority logic and ZSET operations.
Out of Scope Changes check	✅ Passed	All changes are directly scoped to issue #522: PriorityZSetQueue and PriorityZSetConsumer implementations, ZSetIntake worker, parser worker RetryScheduler integration, main.go feature flag wiring, and supporting unit tests with no extraneous modifications.

_{✏️ 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/#522/parser-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.}

[gemini-code-assist]

Code Review

This pull request introduces a Redis ZSET-based priority queueing mechanism for the parser stage to enable priority-based processing that bypasses Kafka's partition FIFO constraints. Key changes include the addition of a ZSetIntake component for transferring messages from Kafka to Redis, a PriorityZSetQueue implementation, and logic within the worker to utilize a RetryScheduler for preventing message loss. Feedback focuses on a critical priority inversion bug caused by an insufficient priorityFactor, hardcoded target types in retry jobs that could break category processing, and the need for better consistency in crawler name resolution and code reuse.

[coderabbitai]

🧹 Nitpick comments (1)

internal/processor/parser/worker/worker.go (1)

328-343: 💤 Low value

Consider logging the original error when retry enqueue fails.

When enqueueRetry fails (line 331-333), the log message includes enqueueErr but not the original processing error err. In ZSET mode, the message will be lost without a record of why it originally failed.

💡 Proposed enhancement for debugging visibility

 		if w.retryScheduler != nil {
 			if enqueueErr := w.enqueueRetry(ctx, msg, err); enqueueErr != nil {
-				log.WithError(enqueueErr).WithField("offset", msg.Offset).Warn("retry enqueue failed, message will be lost in zset mode")
+				log.WithError(enqueueErr).WithFields(map[string]interface{}{
+					"offset":         msg.Offset,
+					"original_error": err.Error(),
+				}).Warn("retry enqueue failed, message will be lost in zset mode")
 				return
 			}

🤖 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/parser/worker/worker.go` around lines 328 - 343, When
enqueueRetry fails in the w.retryScheduler branch, the warning currently logs
only enqueueErr and not the original processing error (err); update the log call
in the failing branch inside the if w.retryScheduler != nil block so it includes
both enqueueErr and the original err (and keep the msg.Offset field) to preserve
the root cause for ZSET-mode message loss debugging; locate the enqueueRetry
call and its enclosing warning log and add the original err to the log context
(e.g., WithError(err) or equivalent) while retaining the existing enqueueErr
detail.

🤖 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.

Nitpick comments:
In `@internal/processor/parser/worker/worker.go`:
- Around line 328-343: When enqueueRetry fails in the w.retryScheduler branch,
the warning currently logs only enqueueErr and not the original processing error
(err); update the log call in the failing branch inside the if w.retryScheduler
!= nil block so it includes both enqueueErr and the original err (and keep the
msg.Offset field) to preserve the root cause for ZSET-mode message loss
debugging; locate the enqueueRetry call and its enclosing warning log and add
the original err to the log context (e.g., WithError(err) or equivalent) while
retaining the existing enqueueErr detail.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 2c18c4f7-1e13-46de-bf5f-f5b7de2f3c2c

📥 Commits

Reviewing files that changed from the base of the PR and between c788305 and 134b3db.

📒 Files selected for processing (7)

• cmd/issuetracker/main.go
• internal/processor/parser/stage.go
• internal/processor/parser/worker/worker.go
• internal/processor/parser/worker/zset_intake.go
• pkg/queue/priority_zset.go
• test/internal/processor/parser/worker/retry_job_test.go
• test/pkg/queue/priority_zset_test.go

[Copilot]

Pull request overview

이 PR은 Parser stage에서 Kafka partition FIFO만으로는 제공하기 어려운 priority sub-ordering을 보장하기 위해, Redis ZSET intermediate queue를 도입하고(feature flag로 opt-in), 실패 시 RetryScheduler 경유 재발행 흐름을 추가합니다. 전체적으로 Phase 2 Parser stage의 처리 우선순위 일관성을 다중 인스턴스 환경까지 확장하는 변경입니다.

Changes:

• pkg/queue.PriorityZSetQueue + PriorityZSetConsumer 추가(푸시/팝/길이, workerpool 재사용)
• Parser stage에 Kafka→ZSET intake goroutine 추가 및 Worker의 RetryScheduler 기반 실패 처리 훅 추가
• cmd/issuetracker/main.go에 feature flag/env wiring 및 관련 단위/통합 테스트 추가

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
pkg/queue/priority_zset.go	Redis ZSET 기반 priority 큐/Consumer 구현 추가
internal/processor/parser/worker/zset_intake.go	Kafka 메시지 intake 후 ZSET 적재 + Kafka commit 루프 추가
internal/processor/parser/worker/worker.go	ProcessMessage 실패 시 RetryScheduler enqueue 경로 및 BuildRetryJob 추가
internal/processor/parser/stage.go	Stage lifecycle에 ZSetIntake 실행(go routine) 통합
cmd/issuetracker/main.go	PARSER_PRIORITY_QUEUE_ENABLED 기반 ZSET 모드 wiring, env helper 추가
test/pkg/queue/priority_zset_test.go	PriorityZSetQueue/Consumer Redis 통합 테스트 추가
test/internal/processor/parser/worker/retry_job_test.go	BuildRetryJob/PriorityFromHeader 단위 테스트 추가