[REFAC#285] 통합 PipelineGuard — Scheduler/Publisher 일관 적용 + Category 단명 TTL

[juhy0987]

연관 이슈

• Closes [REFACTOR] 통합 pipeline guard — Scheduler / Publisher 일관 적용 + Category 단명 TTL #285

구현 내용

라이브 검증 (2026-05-06, 1h 53m) 의 url already in pipeline 11,391건 노이즈 분석 결과, parser 가 카테고리 페이지에서 article URL 을 추출해 publish 할 때 24h IngestionLock 이 정상 차단하지만 publish 시도 자체 (정규화 / Gate / Redis SETNX) 비용이 발생. 또한 시나리오 C (카테고리 URL 이 article 로 잘못 추출) 도 동일 경로로만 차단.

본 PR 은 IngestionLock 의 적용 범위를 모든 publish 진입점으로 확장 + Category 에 단명 TTL 도입 — Scheduler 의 정기 갱신 의도는 보존.

변경점 8가지

#	위치	내용
1	internal/locks/ingestion_lock.go	IngestionLock 인터페이스에 AcquireWithTTL 추가 (호출별 TTL override). RedisIngestionLock / NoopIngestionLock 양쪽 구현
2	internal/locks/pipeline_guard.go (신규)	PipelineGuard 신설. Category=단명 TTL / Article=default TTL. Release 메소드
3	pkg/config/config.go	PIPELINE_GUARD_CATEGORY_TTL 환경변수 (default 60s)
4	internal/scheduler/emitter.go	JobEmitter.SetGuard + Emit 직전 CheckAndAcquire
5	internal/publisher/publisher.go	SetPipelineGuard 추가. guard 우선, IngestionLock fallback (backward compat). Category 우회 제거
6	internal/processor/parser/worker/parser_worker.go	SetPipelineGuard + processCategoryPage defer 로 release 호출
7	cmd/issuetracker/main.go	pipelineGuard 구성 후 publisher / emitter / pw 모두에 주입
8	.env.example	PIPELINE_GUARD_CATEGORY_TTL 주석 + default

테스트 6건 (test/internal/locks/pipeline_guard_test.go)

• Category 단명 TTL 적용 검증
• Article default TTL 적용 검증
• 중복 acquire false 반환
• Release 후 재진입
• nil lock fallback (Noop)
• 0 TTL fallback to DefaultCategoryTTL

기대 효과 (라이브 검증 기준)

항목	현재	적용 후
url already in pipeline Debug 로그	11,391건	거의 0 (publish 시도 전 차단)
Publisher 호출 비용 (정규화 / Gate / Redis)	11,391회 발생	시도 자체 차단
카테고리 cycle overlap	가능 (lock 우회)	차단
Scheduler 정기 갱신 (2h 주기)	정상	동일 (TTL 60s 라 영향 없음)

CI / 머지 게이트 점검

CI 운영 규약 및 Required Status Checks 단일 소스에 따라 작성합니다.

변경 영향 범위

• 영향 패키지: internal/locks, internal/scheduler, internal/publisher, internal/processor/parser/worker, pkg/config, cmd/issuetracker
• 위험도: Medium
  • PipelineGuard 미주입 시 기존 IngestionLock fallback 으로 backward compat 보장
  • Redis 일시 장애 시 fail-open (publish 진행)
  • Release 실패는 non-fatal (TTL fallback)

Required Status Checks

• 통과 확인 대상:
  • Commit Lint
  • PR Title Lint
  • Linked Issue Check
  • Build
  • Test
  • Lint

로컬 검증

• make fmt 통과
• make build 5개 binary 모두 통과
• go test -race ./... 전체 통과 (신규 6건 포함)
• go vet ./... 통과

롤백 계획

• 모든 새 메소드는 backward compat — guard 미주입 시 기존 동작 유지
• 환경변수 미설정 시 default 60s 적용 (운영 영향 없음)
• 단순 revert 만으로 즉시 롤백 가능

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features

  • Pipeline processing now includes coordinated handling mechanisms across all processing stages
  • Improved system resilience with automatic fallback mechanisms for critical operations

• Configuration

  • Added PIPELINE_GUARD_CATEGORY_TTL environment variable (default: 60s) for configuring category processing behavior

• Tests

  • Comprehensive test coverage added for pipeline coordination and fallback scenarios

[coderabbitai]

Warning

Rate limit exceeded

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

To continue reviewing without waiting, purchase usage credits 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: 619832f7-fba7-42a7-974e-94a7954bab68

📥 Commits

Reviewing files that changed from the base of the PR and between 3870d88 and c97af83.

📒 Files selected for processing (6)

• cmd/issuetracker/main.go
• internal/locks/pipeline_guard.go
• internal/processor/parser/worker/parser_worker.go
• internal/publisher/publisher.go
• internal/scheduler/emitter.go
• internal/scheduler/scheduler.go

📝 Walkthrough

Walkthrough

This PR implements a unified PipelineGuard mechanism that enforces target-type-specific TTL semantics for pipeline entry markers. The guard wraps the existing IngestionLock to prevent concurrent pipeline cycles and is wired into the scheduler, publisher, and parser worker to coordinate marker acquisition and release across pipeline stages.

Changes

Unified Pipeline Guard Implementation

Layer / File(s)	Summary
Interface Extension internal/locks/ingestion_lock.go	IngestionLock interface adds AcquireWithTTL method. RedisIngestionLock and NoopIngestionLock implementations provided with TTL fallback behavior when non-positive TTLs are supplied.
Guard Abstraction internal/locks/pipeline_guard.go	New PipelineGuard wrapper enforces target-type-specific TTLs: DefaultCategoryTTL (60s) for categories, underlying lock's default (24h) for articles. CheckAndAcquire acquires markers with appropriate TTLs; Release invalidates markers. Nil-safety fallbacks included.
Configuration pkg/config/config.go, .env.example	RedisConfig adds PipelineGuardCategoryTTL field (default 60s). LoadRedis parses PIPELINE_GUARD_CATEGORY_TTL environment variable.
Publisher Integration internal/publisher/publisher.go	Atomic guard field added; SetPipelineGuard method enables runtime injection. Publish prioritizes guard via acquireViaGuard when configured; falls back to IngestionLock for non-Category targets when guard is unset. Fail-open error handling preserved.
Scheduler Integration internal/scheduler/emitter.go	JobEmitter gains optional guard field and SetGuard method. Emit calls guard.CheckAndAcquire before marshalling/publishing; skips with debug log if acquisition fails (in-progress cycle detected), logs warning on guard errors and proceeds (fail-open).
Parser Worker Integration internal/processor/parser/worker/parser_worker.go	ParserWorker gains optional guard field and SetPipelineGuard method. processCategoryPage defers a releaseCategoryMarker call using timeout and context.WithoutCancel to ensure release on all exit paths; release errors logged as non-fatal.
Main Wiring cmd/issuetracker/main.go	pipelineGuard instantiated when ingestionLock available; injected into jobPublisher, parser worker, and scheduler emitter. Prior ingestion-lock wiring on publisher removed in favor of guard-based dedup.
Tests & Doubles test/internal/locks/pipeline_guard_test.go, test/internal/publisher/publisher_ingestion_lock_test.go	Comprehensive guard tests: category/article TTL selection, duplicate acquisition, release-enable-reacquire, nil-lock fallback, zero-TTL default. Test double fakeIngestionLock extended with AcquireWithTTL method.

Sequence Diagram

sequenceDiagram
    participant Scheduler as Scheduler<br/>(emit)
    participant Guard as PipelineGuard
    participant Lock as IngestionLock
    participant Publisher as Publisher
    participant Parser as Parser Worker
    participant Fetcher as Fetcher Worker

    Scheduler->>Guard: CheckAndAcquire(url, Category)<br/>targetType=Category
    activate Guard
    Guard->>Lock: AcquireWithTTL(url, 60s)<br/>short TTL for category
    Lock-->>Guard: acquired=true
    deactivate Guard
    Guard-->>Scheduler: acquired=true
    
    Scheduler->>Publisher: Emit → Kafka
    
    Publisher->>Guard: CheckAndAcquire(url, Article)<br/>targetType=Article  
    Guard->>Lock: Acquire(url)<br/>uses default 24h TTL
    Lock-->>Guard: acquired=true
    Guard-->>Publisher: acquired=true
    
    Publisher->>Publisher: publish message
    Publisher-->>Fetcher: job → queue

    Fetcher->>Fetcher: fetch & parse
    Fetcher->>Parser: extracted links
    
    Parser->>Guard: CheckAndAcquire(url, Article)<br/>for extracted article
    Guard->>Lock: Acquire(url)
    Lock-->>Guard: acquired=false<br/>(same URL in pipeline)
    Guard-->>Parser: acquired=false
    Parser-->>Parser: skip duplicate
    
    Parser->>Parser: category cycle complete
    Parser->>Guard: Release(url)
    Guard->>Lock: Invalidate(url)
    Lock-->>Guard: done
    
    Note over Scheduler,Fetcher: Category marker released → next<br/>scheduler cycle can re-enter

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Possibly related PRs

• EinSofINTEREST/IssueTracker#202: Both PRs enhance lock infrastructure—this PR builds on prior ingestion-lock refactoring by adding AcquireWithTTL capability and wrapping it in a new PipelineGuard abstraction with target-type semantics.
• EinSofINTEREST/IssueTracker#179: Both PRs modify ingestion-lock dedup flow and publisher wiring—[FEAT#178] Ingestion Lock — URLCache 폐기 + Publisher 단일 책임 정규화 + atomic SETNX 진입 차단 #179 introduces the IngestionLock interface and publisher gating, while this PR extends it with PipelineGuard (TTL flexibility, category release callbacks) and multi-component coordination.

Suggested labels

refactor

Poem

A guard now stands at pipeline gates,
With short TTLs for categories' fates,
When cycles finish, markers release,
Duplicates cease—at last, sweet peace! 🐰✨

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly references issue #285 and accurately summarizes the key changes: introducing PipelineGuard with consistent application across Scheduler/Publisher and short TTL for Categories.
Linked Issues check	✅ Passed	The PR fully implements all coding objectives from issue #285: AcquireWithTTL interface, PipelineGuard mechanism with target-type TTL semantics, category Release callbacks, scheduler/publisher/parser integration, config support, and comprehensive unit tests.
Out of Scope Changes check	✅ Passed	All changes are directly scoped to implementing PipelineGuard (issue #285): adding lock interfaces, creating the guard mechanism, wiring into entry points, and adding config/tests. No unrelated modifications detected.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

_{✏️ 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 refactor/#285/pipeline-guard

Tip

💬 Introducing Slack Agent: The best way for teams to turn conversations into code.

Slack Agent is built on CodeRabbit's deep understanding of your code, so your team can collaborate across the entire SDLC without losing context.

• Generate code and open pull requests
• Plan features and break down work
• Investigate incidents and troubleshoot customer tickets together
• Automate recurring tasks and respond to alerts with triggers
• Summarize progress and report instantly

Built for teams:

• Shared memory across your entire org—no repeating context
• Per-thread sandboxes to safely plan and execute work
• Governance built-in—scoped access, auditability, and budget controls

One agent for your entire SDLC. Right inside Slack.

👉 Get started

---

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]

Pull request overview

이 PR은 IngestionLock을 기반으로 한 통합 PipelineGuard를 도입해 Scheduler/Publisher/ParserWorker의 publish 진입점을 일관되게 보호하고, Category 타겟에 단명 TTL + 명시적 Release를 적용하여 “url already in pipeline” 노이즈 및 불필요한 publish 시도 비용을 줄이려는 리팩토링입니다.

Changes:

• IngestionLock에 AcquireWithTTL을 추가하고, 이를 래핑하는 internal/locks.PipelineGuard를 신설해 Category(단명 TTL) / Article(기본 TTL) 정책을 통합
• Scheduler emitter / Publisher / ParserWorker에 guard 주입 및 (Category) Release 연동
• PIPELINE_GUARD_CATEGORY_TTL 설정 추가 및 PipelineGuard 단위 테스트 추가

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
test/internal/publisher/publisher_ingestion_lock_test.go	publisher용 fake lock이 새 AcquireWithTTL 시그니처를 만족하도록 보강
test/internal/locks/pipeline_guard_test.go	PipelineGuard의 Category TTL/Article TTL/중복 acquire/release/noop fallback 동작을 단위 테스트로 추가
pkg/config/config.go	PIPELINE_GUARD_CATEGORY_TTL 설정을 RedisConfig에 추가하고 환경변수 파싱 로직 반영
internal/scheduler/emitter.go	Emit 직전에 guard 체크를 수행하도록 추가 (중복 cycle 시 emit skip)
internal/publisher/publisher.go	PipelineGuard 우선 적용 + IngestionLock fallback 구조로 publish 진입 dedup 로직 확장
internal/processor/parser/worker/parser_worker.go	Category 처리 종료 시 guard Release를 defer로 보장하도록 추가
internal/locks/pipeline_guard.go	(신규) target type별 TTL 정책을 적용하는 PipelineGuard 구현 추가
internal/locks/ingestion_lock.go	AcquireWithTTL을 인터페이스/구현체(Redis/Noop)에 추가
cmd/issuetracker/main.go	PipelineGuard 구성 및 publisher/emitter/parser worker에 주입 wiring
.env.example	PIPELINE_GUARD_CATEGORY_TTL 예시/주석 추가

[gemini-code-assist]

Code Review

This pull request introduces PipelineGuard, a unified locking mechanism that applies target-specific TTL policies—such as short-lived locks for Categories and 24-hour locks for Articles—across the publisher, scheduler, and parser worker. Key updates include the addition of AcquireWithTTL to the ingestion lock interface and the implementation of explicit lock release in the parser worker for category cycles. Feedback highlights a potential issue with missing URL normalization in the scheduler's emitter, which could lead to inconsistent lock keys compared to the publisher, and suggests using a switch statement in the guard logic to improve extensibility and safety for future target types.

[coderabbitai]

Actionable comments posted: 2

Caution

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

⚠️ Outside diff range comments (1)

internal/publisher/publisher.go (1)

64-71: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Struct doc comment is stale after this PR.

Line 69 states TargetTypeCategory 는 lock 미적용 but when PipelineGuard is active (the new primary path), Category URLs are acquired via CheckAndAcquire with the short-lived TTL. The "no lock for Category" wording now only applies to the legacy IngestionLock fallback path (line 194).

✏️ Suggested doc update

-  //   - TargetTypeCategory 는 lock 미적용 (카테고리 페이지는 매 주기 새 기사 추출이 목적)
+  //   - TargetTypeCategory 는 PipelineGuard 경로에서 단명 TTL (default 60s) 적용;
+  //     IngestionLock fallback 경로에서만 lock 미적용 (backward compat)

🤖 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/publisher/publisher.go` around lines 64 - 71, Update the struct doc
comment to reflect current behavior: change the sentence that says
"TargetTypeCategory 는 lock 미적용" to clarify that when PipelineGuard is active
Category URLs are acquired via PipelineGuard.CheckAndAcquire with a short-lived
TTL, and that "no lock for Category" only applies to the legacy IngestionLock
fallback path (IngestionLock and its behavior at the old fallback code path
remains unchanged); mention that lock lookup failures remain fail-open. Locate
references to PipelineGuard, CheckAndAcquire, and IngestionLock in the comment
and replace the stale statement with this concise, accurate description.

🤖 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 `@internal/publisher/publisher.go`:
- Around line 41-44: The publisher.IngestionLock interface has an unused method
AcquireWithTTL which forces mocks/implementations (e.g., fakeIngestionLock,
SetIngestionLock users) to implement an unnecessary breaking change; remove
AcquireWithTTL from the publisher.IngestionLock interface, update any
tests/mocks that implemented it, and drop the now-unused time import from
internal/publisher/publisher.go; keep behavior that acquireIngestion and callers
(which use lock.Acquire and PipelineGuard.CheckAndAcquire /
internal/locks.IngestionLock) remain unchanged.

In `@internal/scheduler/emitter.go`:
- Around line 47-64: When Emit acquires a guard via e.guard.CheckAndAcquire but
producer.Publish later fails, the guard marker is never released causing
CheckAndAcquire to block retries; add a Release(ctx, url, targetType) method to
the PipelineGuard interface (locks.PipelineGuard already implements Release)
and, inside Emit after a failed producer.Publish call, call e.guard.Release(ctx,
job.Target.URL, job.Target.Type) before returning the publish error so the guard
is cleared and scheduler retries can proceed.

---

Outside diff comments:
In `@internal/publisher/publisher.go`:
- Around line 64-71: Update the struct doc comment to reflect current behavior:
change the sentence that says "TargetTypeCategory 는 lock 미적용" to clarify that
when PipelineGuard is active Category URLs are acquired via
PipelineGuard.CheckAndAcquire with a short-lived TTL, and that "no lock for
Category" only applies to the legacy IngestionLock fallback path (IngestionLock
and its behavior at the old fallback code path remains unchanged); mention that
lock lookup failures remain fail-open. Locate references to PipelineGuard,
CheckAndAcquire, and IngestionLock in the comment and replace the stale
statement with this concise, accurate description.

🪄 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: f0689c41-6330-4c31-b268-c032d7d1a7be

📥 Commits

Reviewing files that changed from the base of the PR and between 936277c and 3870d88.

📒 Files selected for processing (10)

• .env.example
• cmd/issuetracker/main.go
• internal/locks/ingestion_lock.go
• internal/locks/pipeline_guard.go
• internal/processor/parser/worker/parser_worker.go
• internal/publisher/publisher.go
• internal/scheduler/emitter.go
• pkg/config/config.go
• test/internal/locks/pipeline_guard_test.go
• test/internal/publisher/publisher_ingestion_lock_test.go