[FEAT#523] Validate stage Redis ZSET intermediate queue — priority sub-ordering

[juhy0987]

연관 이슈

• Closes [FEATURE] Validate stage priority-aware 처리 — Redis ZSET intermediate queue #523
• Parent: [FEATURE] host/path 기반 priority 차등 배분 — publish 단계 + parser/validate/enrich 단계 priority-aware 처리 #515 (Phase 2 Sub 5)
• 선행: [FEATURE] Parser stage priority-aware 처리 — Redis ZSET intermediate queue #522 (PR [FEAT#522] Parser stage Redis ZSET intermediate queue — priority sub-ordering #526 머지 완료, pkg/queue.PriorityZSetQueue + PriorityPusher 인터페이스 활용)

구현 내용

메타 이슈 #515 의 Phase 2 Validate stage — Parser #522 와 동일 패턴 적용. Kafka partition FIFO 가 priority sub-ordering 을 제공 못 하는 한계를 Redis ZSET 으로 해소.

흐름:

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

Sub 1 — Validate Worker RetryScheduler hook (6c848ca)

• Worker.retryScheduler bus.RetryScheduler 필드 + SetRetryScheduler setter
• Handle 분기: process 실패 시
  • retryScheduler 주입 시 → enqueueRetry 후 commit (메시지 손실 방지)
  • 미주입 시 → 기존 commit skip → Kafka redeliver (Parser 와 동일)
• BuildRetryJob(msg) (*core.CrawlJob, error): ProcessingMessage → ContentRef → CrawlJob 변환
  • msg.Headers crawler/target_type 우선, ContentRef.SourceInfo fallback
  • PriorityFromHeader 헬퍼로 priority parsing 통일
  • retry_reason="validate_process_failed" + original_ref_id metadata
  • MaxRetries = bus.DefaultMaxRetries

Sub 2 — Kafka → ZSET intake goroutine (d301d92)

• worker.ZSetIntake — Parser 동일 패턴, ContentRef 대응
• FetchMessage → ProcessingMessage unmarshal → ContentRef unmarshal → ZSET Push → Kafka commit
• ZSET member key = ContentRef.ID
• 실패 정책 (Parser 와 동일): unmarshal 실패 / 빈 ID → commit / push 실패 → commit skip / commit 실패 → idempotent
• defer consumer.Close() 로 Kafka reader 자원 정리

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

• VALIDATE_PRIORITY_QUEUE_ENABLED=true + redisClientShared != nil 시 ZSET 모드 활성
• ZSET 모드 + retryScheduler nil → fatal (메시지 손실 방지 가드)
• validate.Stage.SetZSetIntake 로 lifecycle 통합
• 환경변수: VALIDATE_ZSET_QUEUE_KEY / VALIDATE_ZSET_ENTRY_PREFIX / VALIDATE_ZSET_MAX_SIZE / VALIDATE_ZSET_ENTRY_TTL / VALIDATE_ZSET_POP_TIMEOUT
• .env.example 갱신

Sub 4 — 단위 테스트 (500a1c1)

• BuildRetryJob 11 cases / PriorityFromHeader 9 cases / ZSetIntake.handleOne 6 cases
• 모두 Redis / Kafka 의존 없는 순수 단위

CI / 머지 게이트 점검

변경 영향 범위

• 영향 패키지/모듈: internal/processor/validate, cmd/issuetracker, .env.example
• 위험도: Medium — Validate 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. VALIDATE_PRIORITY_QUEUE_ENABLED=false (또는 unset) — 기존 Kafka consume 흐름 복귀
2. ZSET 잔존 항목은 entry TTL 24h 후 자연 정리
3. 영구 롤백: 위 환경변수 + 본 PR revert

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

• [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 Redis-backed priority queue for the validate stage with Kafka→ZSET intake.
  • Automatic retry enqueueing for failed validation messages (with DLQ fallback on enqueue failure).

• Configuration

  • New env vars to control priority queue: VALIDATE_PRIORITY_QUEUE_ENABLED, VALIDATE_ZSET_QUEUE_KEY, VALIDATE_ZSET_ENTRY_PREFIX, VALIDATE_ZSET_MAX_SIZE, VALIDATE_ZSET_ENTRY_TTL, VALIDATE_ZSET_POP_TIMEOUT.

• Tests

  • Added unit tests for retry-job construction and ZSET intake handling.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 076bcd86-2445-46de-b75a-439859cb4aba

📥 Commits

Reviewing files that changed from the base of the PR and between 500a1c1 and 88ec52e.

📒 Files selected for processing (3)

• cmd/issuetracker/main.go
• internal/processor/validate/worker/worker.go
• test/internal/processor/validate/worker/retry_job_test.go

---

📝 Walkthrough

Walkthrough

Validate stage now supports optional Redis ZSET-based intake for priority-aware message processing. Kafka messages are conditionally bridged into a ZSET, workers pop by priority, and failed validations are retried by reconstructing job metadata from message headers.

Changes

Validate ZSET Intake and Retry Flow

Layer / File(s)	Summary
Configuration and environment variables .env.example	Redis ZSET queue settings added: enablement flag, queue key, entry prefix, max size, TTL, and pop timeout.
Stage lifecycle wiring for ZSET intake internal/processor/validate/stage.go	Stage gains optional intake *worker.ZSetIntake and SetZSetIntake; Start conditionally launches intake.Run(ctx) when set.
ZSET intake component (Kafka→Redis bridge) internal/processor/validate/worker/zset_intake.go	ZSetIntake polls Kafka, unmarshals ProcessingMessage/ContentRef, computes priority, pushes to PriorityZSetQueue, and commits Kafka on push success; parse/schema/empty-ID errors commit to avoid redelivery loops; push failures skip commit.
ZSetIntake unit tests and stubs test/internal/processor/validate/worker/zset_intake_test.go	Stub PriorityPusher and Consumer validate HandleOne behavior: success push+commit, unmarshal failures commit, empty ID commits, push failure skips commit, and default priority mapping.
Bootstrap and main.go orchestration cmd/issuetracker/main.go	Creates dedicated validate Kafka consumer, conditionally builds PriorityZSetQueue/PriorityZSetConsumer when enabled, creates ZSetIntake to bridge Kafka→ZSET, injects retryScheduler in ZSET mode (fatal if missing), and wires intake into stage via SetZSetIntake.
Worker retry scheduler and retry-job construction internal/processor/validate/worker/worker.go	Worker gains retryScheduler field and SetRetryScheduler. Handle now enqueues a retry CrawlJob via BuildRetryJob and RetryScheduler.Enqueue on processing errors (with DLQ fallback), and commits original offset on enqueue success. PriorityFromHeader and isValidTargetType added.
BuildRetryJob and PriorityFromHeader tests test/internal/processor/validate/worker/retry_job_test.go	Tests cover valid retry job construction, priority and timeout header handling/defaults, crawler header precedence, target type parsing, and malformed payload error paths; table-driven tests for priority mapping/defaulting.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• EinSofINTEREST/IssueTracker#526: Implements parallel ZSET intake + retry scheduler pattern for parser stage using identical BuildRetryJob/PriorityFromHeader/ZSetIntake approach.
• EinSofINTEREST/IssueTracker#207: Introduced the original processor.Stage and validate.Stage wrapper structure that this PR extends with ZSetIntake field and lifecycle management.
• EinSofINTEREST/IssueTracker#418: Previously refactored validate.Stage wiring and constructor signatures that this PR further extends with optional ZSET intake.

Suggested labels

enhancement

Poem

🐰 I hopped from Kafka to Redis with glee,
Pushed jobs by priority, one-two-three,
When bumps arrive, I stitch the job anew,
RetryScheduler pats each failed queue,
A tiny rabbit keeps the validate tree.

🚥 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#523] Validate stage Redis ZSET intermediate queue — priority sub-ordering' clearly and specifically describes the main change: adding a Redis ZSET-based intermediate queue to the Validate stage for priority-based sub-ordering.
Linked Issues check	✅ Passed	The PR successfully implements all primary coding objectives from issue #523: Kafka→ZSET intake with priority calculation and Kafka commit, ZSET pop for priority-ordered processing, retry scheduling on failures, environment variable configuration, feature flag gating, and comprehensive unit tests covering retry jobs, priority header parsing, and ZSET intake logic.
Out of Scope Changes check	✅ Passed	All changes are directly aligned with issue #523 requirements: .env.example updates for ZSET configuration, intake goroutine implementation, worker retry mechanism, stage lifecycle integration, and unit test coverage. No extraneous or unrelated modifications detected.

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

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feature/#523/validate-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 implements a priority queue for the validation stage using Redis ZSET, enabling priority-based sub-ordering of messages by moving them from Kafka to an intermediate Redis queue. Key changes include the introduction of a ZSetIntake component to handle the message transfer, lifecycle management for this process within the validation stage, and the integration of a RetryScheduler to prevent message loss during processing failures in ZSET mode. Feedback suggests improving the BuildRetryJob function by inheriting the timeout value from message headers instead of using a hardcoded 30-second default, which would ensure consistency with existing republishing logic.

[Copilot]

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

[coderabbitai]

Actionable comments posted: 2

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

Inline comments:
In `@cmd/issuetracker/main.go`:
- Around line 1052-1055: The validate-only initialization is being enabled even
when the Validate stage is globally disabled; update the gating logic so
validatePriorityQueueEnabled also requires stagesCfg.ValidateEnabled (i.e.,
combine envBoolOrDefault("VALIDATE_PRIORITY_QUEUE_ENABLED", false) &&
redisClientShared != nil && stagesCfg.ValidateEnabled) and apply the same change
to the duplicate block around validateKafkaConsumer/validateZSetIntake (the
other occurrence referenced in the review), ensuring both
validatePriorityQueueEnabled declarations and any conditionals that instantiate
validateKafkaConsumer or validateWorkerPkg.ZSetIntake honor
stagesCfg.ValidateEnabled.

In `@internal/processor/validate/worker/worker.go`:
- Around line 168-172: The current block that calls w.enqueueRetry(ctx, msg,
err) (via retryScheduler) can drop the message if enqueueErr occurs; modify this
branch so that when enqueueErr != nil you perform a durable fallback before
returning: call the component that publishes to the DLQ (or reinsert into the
durable queue) with the original msg and error metadata (use the same message
payload and include err details), log any errors from that fallback, and only
then return; update the code around retryScheduler / enqueueRetry to invoke the
DLQ publish method (or durableQueue.Enqueue) as the fallback so messages are not
lost when retry enqueue fails.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: ed7c5fad-5193-4381-b214-0075d44ab104

📥 Commits

Reviewing files that changed from the base of the PR and between 1b4769f and 500a1c1.

📒 Files selected for processing (7)

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