From 91edc206eb08bb66ce505ba4d456c56d4e6c8c61 Mon Sep 17 00:00:00 2001 From: juhyuni Date: Mon, 18 May 2026 22:59:44 +0900 Subject: [PATCH 1/7] =?UTF-8?q?[FEAT]:=20Redis=20LIST=20=EA=B8=B0=EB=B0=98?= =?UTF-8?q?=20JobBuffer=20=EC=B6=94=EA=B0=80=20=E2=80=94=20pkg/redis=20?= =?UTF-8?q?=EB=8B=A8=EC=9D=BC=20=EC=B1=85=EC=9E=84=20(=EC=9D=B4=EC=8A=88?= =?UTF-8?q?=20#510)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pkg/redis/job_buffer.go: EnqueueJob / DrainJobs / JobBufferLen - LIST 자료구조: LPUSH 로 head enqueue, RPOP COUNT 로 tail FIFO drain (Redis 6.2+) - MaxLen 전달 시 LTRIM 으로 oldest 제거 (buffer 무한 누적 방어) - 기존 retry_queue.go (ZSET) 와 동일 패키지 컨벤션 — Client 메소드 + JobBufferKey 헬퍼 - 단위 테스트 5개 (FIFO / MaxLen LTRIM / empty / validation / Len) — Redis 환경에서 통합 검증 Co-Authored-By: Claude Opus 4.7 (1M context) --- pkg/redis/job_buffer.go | 113 ++++++++++++++++++++++++++++ test/pkg/redis/job_buffer_test.go | 120 ++++++++++++++++++++++++++++++ 2 files changed, 233 insertions(+) create mode 100644 pkg/redis/job_buffer.go create mode 100644 test/pkg/redis/job_buffer_test.go diff --git a/pkg/redis/job_buffer.go b/pkg/redis/job_buffer.go new file mode 100644 index 00000000..be761edf --- /dev/null +++ b/pkg/redis/job_buffer.go @@ -0,0 +1,113 @@ +package redis + +import ( + "context" + "errors" + "fmt" + + goredis "github.com/redis/go-redis/v9" +) + +// JobBufferKeyPrefix 는 normal/low priority crawl job 의 직렬화 Kafka payload 를 임시 적재하는 +// Redis LIST 키의 공통 접두사입니다 (이슈 #510). 실제 키는 publisher:buffer: 형식. +// +// 자료구조 — LIST (FIFO): +// - LPUSH 로 head 에 enqueue +// - RPOP COUNT N 로 tail 에서 drain (가장 오래된 항목부터) +// - LLEN 으로 모니터링 +const JobBufferKeyPrefix = "publisher:buffer:" + +// JobBufferKey 는 priority label 에 대응하는 buffer LIST 키를 반환합니다. +// label 은 호출자 컨벤션 — 일반적으로 "normal" / "low". +func JobBufferKey(label string) string { + return JobBufferKeyPrefix + label +} + +// EnqueueJob 은 payload 를 priority label 버퍼의 head 에 추가합니다. +// +// MaxLen > 0 이면 LPUSH 후 LTRIM 으로 LIST 길이를 MaxLen 이하로 보정 — buffer 무한 누적을 방어. +// MaxLen <= 0 이면 길이 제한 없음 (운영 cautious 시 0 가능, 단 metric 으로 모니터링 필수). +// +// LTRIM 은 oldest 부터 제거 — buffer 가 임계 도달 시 가장 오래된 (= drain 기회를 가장 오래 못 받은) +// 항목이 소실되지만, 정책상 신규 publish 우선. 호출자는 enqueue 직전 IngestionLock 로 dedup 책임. +// +// label / payload 빈 값은 명시적 error — silent corruption 회피. +func (c *Client) EnqueueJob(ctx context.Context, label string, payload []byte, maxLen int64) error { + if label == "" { + return errors.New("enqueue job: empty label") + } + if len(payload) == 0 { + return fmt.Errorf("enqueue job %s: empty payload", label) + } + + key := JobBufferKey(label) + + if maxLen > 0 { + // LPUSH + LTRIM 을 pipeline 으로 1 RTT — 두 명령 사이 race 가 있어도 다음 LTRIM 에서 보정. + pipe := c.rdb.Pipeline() + pipe.LPush(ctx, key, payload) + // LTRIM start stop 은 inclusive — [0, maxLen-1] 보존 (head 최신 N개). + pipe.LTrim(ctx, key, 0, maxLen-1) + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("enqueue job %s: %w", label, err) + } + return nil + } + + if err := c.rdb.LPush(ctx, key, payload).Err(); err != nil { + return fmt.Errorf("enqueue job %s: %w", label, err) + } + return nil +} + +// DrainJobs 는 priority label 버퍼의 tail 에서 최대 n 개의 payload 를 pop 하여 반환합니다. +// +// RPOP COUNT n (Redis 6.2+) 으로 1 RTT 에 일괄 처리 — N 회 round-trip 회피. +// 반환 순서: tail 우선 (= 가장 오래된 순) → FIFO 보장. +// +// n <= 0 이면 빈 슬라이스 반환. +// LIST 가 비어 있으면 빈 슬라이스 반환 (error 아님 — 정상 idle). +// +// **at-most-once 의미**: drain 된 항목은 즉시 Redis 에서 제거됨. drainer 가 Kafka publish 에 +// 실패한 경우 호출자가 다시 EnqueueJob 으로 재적재 — peek-publish-ack 가 아닌 pop-publish 패턴 +// (단순성 우선). publish 실패 시 호출자가 재적재 책임. +// +// 다중 인스턴스 race: 동일 LIST 에서 RPOP 은 atomic — 같은 항목이 두 인스턴스에 동시 반환 안 됨. +// 두 drainer 인스턴스가 동작해도 각자 다른 항목 처리. +func (c *Client) DrainJobs(ctx context.Context, label string, n int) ([][]byte, error) { + if n <= 0 { + return nil, nil + } + if label == "" { + return nil, errors.New("drain jobs: empty label") + } + + key := JobBufferKey(label) + + res, err := c.rdb.RPopCount(ctx, key, n).Result() + if errors.Is(err, goredis.Nil) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("drain jobs %s: %w", label, err) + } + + out := make([][]byte, 0, len(res)) + for _, s := range res { + out = append(out, []byte(s)) + } + return out, nil +} + +// JobBufferLen 은 priority label 버퍼의 현재 길이를 반환합니다 (모니터링/디버깅용). +// 키가 부재하면 (0, nil) 반환. +func (c *Client) JobBufferLen(ctx context.Context, label string) (int64, error) { + if label == "" { + return 0, errors.New("job buffer len: empty label") + } + n, err := c.rdb.LLen(ctx, JobBufferKey(label)).Result() + if err != nil { + return 0, fmt.Errorf("llen %s: %w", JobBufferKey(label), err) + } + return n, nil +} diff --git a/test/pkg/redis/job_buffer_test.go b/test/pkg/redis/job_buffer_test.go new file mode 100644 index 00000000..0b86d58d --- /dev/null +++ b/test/pkg/redis/job_buffer_test.go @@ -0,0 +1,120 @@ +package redis_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + pkgredis "issuetracker/pkg/redis" +) + +// jobBufferCleanup 은 테스트 간 격리를 위해 buffer 의 모든 항목을 drain 합니다. +func jobBufferCleanup(t *testing.T, client *pkgredis.Client, label string) { + t.Helper() + ctx := context.Background() + for { + drained, err := client.DrainJobs(ctx, label, 1000) + require.NoError(t, err) + if len(drained) == 0 { + return + } + } +} + +// TestEnqueueDrainJob_BasicFIFO 는 enqueue 순서대로 drain 되는지 (FIFO) 검증합니다. +func TestEnqueueDrainJob_BasicFIFO(t *testing.T) { + client := newTestClient(t) + label := "test-fifo" + jobBufferCleanup(t, client, label) + ctx := context.Background() + + for i := 0; i < 5; i++ { + payload := []byte("payload-" + string(rune('a'+i))) + require.NoError(t, client.EnqueueJob(ctx, label, payload, 0)) + } + + drained, err := client.DrainJobs(ctx, label, 5) + require.NoError(t, err) + require.Len(t, drained, 5) + + // FIFO 검증 — 첫 enqueue (payload-a) 가 첫 drain 위치에 등장. + for i, p := range drained { + expected := "payload-" + string(rune('a'+i)) + assert.Equal(t, expected, string(p), "FIFO 순서 — index %d", i) + } + + // 추가 drain 은 빈 결과. + more, err := client.DrainJobs(ctx, label, 1) + require.NoError(t, err) + assert.Empty(t, more) +} + +// TestEnqueueJob_MaxLenLTrim 은 MaxLen 초과 시 oldest 가 제거되는지 검증합니다. +func TestEnqueueJob_MaxLenLTrim(t *testing.T) { + client := newTestClient(t) + label := "test-maxlen" + jobBufferCleanup(t, client, label) + ctx := context.Background() + + // MaxLen=3 — 5개 enqueue 시 가장 오래된 2개가 LTRIM 으로 제거되어 마지막 3개만 잔존. + for i := 0; i < 5; i++ { + payload := []byte("p-" + string(rune('a'+i))) + require.NoError(t, client.EnqueueJob(ctx, label, payload, 3)) + } + + n, err := client.JobBufferLen(ctx, label) + require.NoError(t, err) + assert.Equal(t, int64(3), n, "MaxLen=3 으로 LTRIM 적용") + + drained, err := client.DrainJobs(ctx, label, 10) + require.NoError(t, err) + require.Len(t, drained, 3) + // FIFO drain — tail (= 가장 오래된 잔존 = p-c) 부터. + assert.Equal(t, "p-c", string(drained[0])) + assert.Equal(t, "p-d", string(drained[1])) + assert.Equal(t, "p-e", string(drained[2])) +} + +// TestDrainJobs_EmptyBuffer 는 빈 buffer 에서 drain 시 빈 슬라이스 + nil error 반환을 검증합니다. +func TestDrainJobs_EmptyBuffer(t *testing.T) { + client := newTestClient(t) + label := "test-empty" + jobBufferCleanup(t, client, label) + ctx := context.Background() + + drained, err := client.DrainJobs(ctx, label, 10) + require.NoError(t, err) + assert.Empty(t, drained) +} + +// TestEnqueueJob_EmptyValidation 은 빈 label / payload 가 error 를 반환하는지 검증합니다. +func TestEnqueueJob_EmptyValidation(t *testing.T) { + client := newTestClient(t) + ctx := context.Background() + + assert.Error(t, client.EnqueueJob(ctx, "", []byte("x"), 0)) + assert.Error(t, client.EnqueueJob(ctx, "ok", nil, 0)) + assert.Error(t, client.EnqueueJob(ctx, "ok", []byte{}, 0)) +} + +// TestJobBufferLen 은 enqueue 후 len 이 정확한지 검증합니다. +func TestJobBufferLen(t *testing.T) { + client := newTestClient(t) + label := "test-len" + jobBufferCleanup(t, client, label) + ctx := context.Background() + + n, err := client.JobBufferLen(ctx, label) + require.NoError(t, err) + assert.Equal(t, int64(0), n) + + for i := 0; i < 7; i++ { + require.NoError(t, client.EnqueueJob(ctx, label, []byte("x"), 0)) + } + + n, err = client.JobBufferLen(ctx, label) + require.NoError(t, err) + assert.Equal(t, int64(7), n) +} From e421cd518336f1fbaa7eb439ff7a49393b680072 Mon Sep 17 00:00:00 2001 From: juhyuni Date: Mon, 18 May 2026 22:59:54 +0900 Subject: [PATCH 2/7] =?UTF-8?q?[FEAT]:=20BufferingProducer=20=EB=8D=B0?= =?UTF-8?q?=EC=BD=94=EB=A0=88=EC=9D=B4=ED=84=B0=20=E2=80=94=20normal/low?= =?UTF-8?q?=20crawl=20=EB=A9=94=EC=8B=9C=EC=A7=80=20Redis=20=EB=9D=BC?= =?UTF-8?q?=EC=9A=B0=ED=8C=85=20(=EC=9D=B4=EC=8A=88=20#510)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pkg/queue/buffering_producer.go: queue.Producer 데코레이터 - topic == TopicCrawlNormal/Low → JobBuffer.EnqueueJob - 그 외 (high / raw / normalized / ...) → underlying.Publish 직접 - buffer enqueue 실패 시 underlying 으로 fallback — Redis 장애가 publish 자체 차단하지 않도록 - PublishBatch 도 메시지 단위 routing (mixed batch 지원) - pkg/queue/buffered_message.go: BufferedAt 기록 포함 JSON 직렬화 + DecodeBufferedMessage (drainer 사용) - NoopJobBuffer / Underlying() 접근자 노출 — drainer 무한 루프 회피용 - 단위 테스트 9개 (mock JobBuffer + Producer) — normal/low/high routing + fallback + nil buffer + Encode/Decode Co-Authored-By: Claude Opus 4.7 (1M context) --- pkg/queue/buffered_message.go | 58 +++++ pkg/queue/buffering_producer.go | 185 ++++++++++++++++ test/pkg/queue/buffering_producer_test.go | 249 ++++++++++++++++++++++ 3 files changed, 492 insertions(+) create mode 100644 pkg/queue/buffered_message.go create mode 100644 pkg/queue/buffering_producer.go create mode 100644 test/pkg/queue/buffering_producer_test.go diff --git a/pkg/queue/buffered_message.go b/pkg/queue/buffered_message.go new file mode 100644 index 00000000..111915dd --- /dev/null +++ b/pkg/queue/buffered_message.go @@ -0,0 +1,58 @@ +package queue + +import ( + "encoding/json" + "fmt" + "time" +) + +// bufferedMessage 는 Redis JobBuffer 에 저장되는 직렬화 가능한 Message 표현입니다 (이슈 #510). +// +// queue.Message 는 map[string]string 헤더와 []byte 필드를 포함 — encoding/json 으로 직렬화 가능. +// 그러나 호환성 / forward-compat 위해 명시적 struct 로 마샬링 — 향후 Message 에 non-serializable +// 필드 추가될 가능성 대비. +type bufferedMessage struct { + Topic string `json:"topic"` + Key []byte `json:"key,omitempty"` + Value []byte `json:"value"` + Time time.Time `json:"time,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + BufferedAt time.Time `json:"buffered_at"` +} + +// encodeBufferedMessage 는 Message 를 JSON 직렬화하여 Redis 저장용 payload 를 생성합니다. +// BufferedAt 은 drainer 가 buffer 체류 시간을 측정할 수 있도록 enqueue 시점에 기록. +func encodeBufferedMessage(msg Message) ([]byte, error) { + bm := bufferedMessage{ + Topic: msg.Topic, + Key: msg.Key, + Value: msg.Value, + Time: msg.Time, + Headers: msg.Headers, + BufferedAt: time.Now(), + } + data, err := json.Marshal(bm) + if err != nil { + return nil, fmt.Errorf("encode buffered message: %w", err) + } + return data, nil +} + +// DecodeBufferedMessage 는 Redis 에서 drain 한 payload 를 Message 로 역직렬화합니다. +// BufferDrainer 가 사용 — drainer 패키지에서 본 함수를 호출해 Kafka publish 입력 구성. +// +// 반환된 bufferedAt 은 drainer 의 buffer 체류 시간 metric / 로그 용도. msg 의 Time 과는 다름 +// (Time 은 producer 가 message 생성 시점, BufferedAt 은 Redis enqueue 시점). +func DecodeBufferedMessage(payload []byte) (Message, time.Time, error) { + var bm bufferedMessage + if err := json.Unmarshal(payload, &bm); err != nil { + return Message{}, time.Time{}, fmt.Errorf("decode buffered message: %w", err) + } + return Message{ + Topic: bm.Topic, + Key: bm.Key, + Value: bm.Value, + Time: bm.Time, + Headers: bm.Headers, + }, bm.BufferedAt, nil +} diff --git a/pkg/queue/buffering_producer.go b/pkg/queue/buffering_producer.go new file mode 100644 index 00000000..4020d856 --- /dev/null +++ b/pkg/queue/buffering_producer.go @@ -0,0 +1,185 @@ +package queue + +import ( + "context" + "errors" + "fmt" + + "issuetracker/pkg/logger" +) + +// JobBuffer 는 crawl topic 메시지를 priority label 별로 임시 적재하는 buffer 인터페이스입니다 +// (이슈 #510). Redis LIST 기반 FIFO 구현체가 pkg/redis 에 존재. +// +// 본 인터페이스는 BufferingProducer 와 BufferDrainer 가 공유 — Redis 직접 의존 회피. +// Noop 구현체 (NoopJobBuffer) 를 wiring 에서 사용하면 buffer 기능이 자동 비활성. +type JobBuffer interface { + EnqueueJob(ctx context.Context, label string, payload []byte, maxLen int64) error + DrainJobs(ctx context.Context, label string, n int) ([][]byte, error) + JobBufferLen(ctx context.Context, label string) (int64, error) +} + +// NoopJobBuffer 는 모든 호출이 zero-value 를 반환하는 fail-safe 구현체입니다. +// Redis 비활성 / buffer 기능 opt-out 시 wiring 에서 사용. +type NoopJobBuffer struct{} + +// EnqueueJob 항상 error 반환 — BufferingProducer 가 fallback 경로를 타도록 신호. +func (NoopJobBuffer) EnqueueJob(_ context.Context, _ string, _ []byte, _ int64) error { + return errors.New("noop job buffer: enqueue not supported") +} + +// DrainJobs 항상 빈 슬라이스 반환 — drainer 가 매 tick 마다 idle 로 인식. +func (NoopJobBuffer) DrainJobs(_ context.Context, _ string, _ int) ([][]byte, error) { + return nil, nil +} + +// JobBufferLen 항상 0 — 모니터링이 buffer 미사용 으로 인식. +func (NoopJobBuffer) JobBufferLen(_ context.Context, _ string) (int64, error) { + return 0, nil +} + +// BufferingProducer 는 normal / low priority crawl topic 메시지를 Redis JobBuffer 에 임시 +// 적재하는 Producer 데코레이터입니다 (이슈 #510). +// +// 동작: +// - msg.Topic == TopicCrawlNormal → JobBuffer.EnqueueJob(label="normal", payload) +// - msg.Topic == TopicCrawlLow → JobBuffer.EnqueueJob(label="low", payload) +// - 그 외 (high crawl / raw / normalized / validated / enriched / dlq / ...) → underlying.Publish +// +// Enqueue 실패 시 underlying producer 로 fallback — Redis 장애가 publish 자체를 막지 않도록 +// (fail-open). 빈도가 잦으면 운영자가 WARN 로그로 감지. +// +// PublishBatch 는 message 단위로 routing — 같은 batch 안에 normal/low/high 가 섞여 있어도 정상 처리 +// 단, Redis enqueue 와 Kafka publish 가 서로 다른 트랜잭션이라 batch atomicity 는 보장 X +// (기존 KafkaProducer.WriteMessages 의 multi-message atomicity 와 동일 한계). +type BufferingProducer struct { + underlying Producer + buffer JobBuffer + maxLen int64 + log *logger.Logger +} + +// NewBufferingProducer 는 underlying Producer 를 buffer 데코레이터로 감쌉니다. +// +// 인자: +// - underlying : 실제 Kafka publish 책임 (보통 *KafkaProducer) +// - buffer : Redis 또는 Noop. nil 이면 NoopJobBuffer 로 자동 대체 — 기능 비활성 +// - maxLen : EnqueueJob 의 LIST 최대 길이 (>0 이면 LTRIM 적용) +// - log : enqueue 실패 / fallback 등 WARN 로그 +// +// underlying / log 는 필수 (nil 이면 NewBufferingProducer 자체가 fallback 으로 underlying 만 반환할 수 +// 있으나, 그러면 데코 의미 없음 → 명시적 error 또는 호출자가 wiring 책임). +func NewBufferingProducer(underlying Producer, buffer JobBuffer, maxLen int64, log *logger.Logger) *BufferingProducer { + if buffer == nil { + buffer = NoopJobBuffer{} + } + return &BufferingProducer{ + underlying: underlying, + buffer: buffer, + maxLen: maxLen, + log: log, + } +} + +// labelForTopic 은 crawl topic → buffer label 매핑을 반환합니다. +// normal/low 외 토픽 (high / 다른 stage 토픽) 은 빈 문자열 — 호출자가 buffer 우회 신호로 사용. +func labelForTopic(topic string) string { + switch topic { + case TopicCrawlNormal: + return "normal" + case TopicCrawlLow: + return "low" + default: + return "" + } +} + +// Publish 는 msg.Topic 에 따라 buffer 또는 underlying 으로 라우팅합니다. +func (p *BufferingProducer) Publish(ctx context.Context, msg Message) error { + label := labelForTopic(msg.Topic) + if label == "" { + return p.underlying.Publish(ctx, msg) + } + + payload, err := encodeBufferedMessage(msg) + if err != nil { + // 직렬화 실패는 underlying 에 직접 위임해 publish 손실 회피. + p.log.WithFields(map[string]interface{}{ + "topic": msg.Topic, + "label": label, + }).WithError(err).Warn("buffered message encode failed, falling back to direct publish") + return p.underlying.Publish(ctx, msg) + } + + if err := p.buffer.EnqueueJob(ctx, label, payload, p.maxLen); err != nil { + // Redis 장애 등 — underlying 으로 fallback. 운영자가 WARN 누적 감지. + p.log.WithFields(map[string]interface{}{ + "topic": msg.Topic, + "label": label, + }).WithError(err).Warn("buffer enqueue failed, falling back to direct publish") + return p.underlying.Publish(ctx, msg) + } + return nil +} + +// PublishBatch 는 메시지 단위로 routing 합니다 — 같은 batch 안 high/normal/low 혼합 가능. +// +// 구현 — buffer-target 과 direct-target 을 두 슬라이스로 분리 후: +// 1. buffer-target: 순차 EnqueueJob (실패 시 direct 로 재분류) +// 2. direct-target: underlying.PublishBatch 1회 +// +// 모든 enqueue 가 성공해도 direct-target 의 PublishBatch 가 실패하면 본 batch 가 부분 성공 — +// 호출자는 본 batch 단위의 transactional 보장을 기대하지 않아야 함 (KafkaProducer 와 동일). +func (p *BufferingProducer) PublishBatch(ctx context.Context, msgs []Message) error { + if len(msgs) == 0 { + return nil + } + + direct := make([]Message, 0, len(msgs)) + for _, msg := range msgs { + label := labelForTopic(msg.Topic) + if label == "" { + direct = append(direct, msg) + continue + } + payload, err := encodeBufferedMessage(msg) + if err != nil { + p.log.WithFields(map[string]interface{}{ + "topic": msg.Topic, + "label": label, + }).WithError(err).Warn("buffered message encode failed, falling back to direct publish") + direct = append(direct, msg) + continue + } + if err := p.buffer.EnqueueJob(ctx, label, payload, p.maxLen); err != nil { + p.log.WithFields(map[string]interface{}{ + "topic": msg.Topic, + "label": label, + }).WithError(err).Warn("buffer enqueue failed, falling back to direct publish") + direct = append(direct, msg) + continue + } + } + + if len(direct) == 0 { + return nil + } + if err := p.underlying.PublishBatch(ctx, direct); err != nil { + return fmt.Errorf("buffering producer direct batch (%d msgs): %w", len(direct), err) + } + return nil +} + +// Close 는 underlying.Close 를 위임 호출합니다. JobBuffer 의 lifecycle 은 외부 (Client) 가 관리. +func (p *BufferingProducer) Close() error { + return p.underlying.Close() +} + +// Underlying 은 BufferDrainer 가 drain 결과를 직접 Kafka 로 publish 할 때 사용할 +// underlying Producer 접근자입니다 (data flow: Redis pop → underlying.PublishBatch). +// +// drainer 는 BufferingProducer 가 아니라 underlying 을 호출해야 함 — 그렇지 않으면 drain 한 +// payload 가 다시 buffer 에 enqueue 되는 무한 루프. +func (p *BufferingProducer) Underlying() Producer { + return p.underlying +} diff --git a/test/pkg/queue/buffering_producer_test.go b/test/pkg/queue/buffering_producer_test.go new file mode 100644 index 00000000..a435c1c1 --- /dev/null +++ b/test/pkg/queue/buffering_producer_test.go @@ -0,0 +1,249 @@ +package queue_test + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "issuetracker/pkg/logger" + "issuetracker/pkg/queue" +) + +// fakeProducer 는 underlying Kafka producer 의 in-memory mock. +type fakeProducer struct { + mu sync.Mutex + published []queue.Message + failErr error // Publish/PublishBatch 가 모두 이 에러 반환 — 실패 시나리오 시뮬레이션 +} + +func (p *fakeProducer) Publish(_ context.Context, msg queue.Message) error { + if p.failErr != nil { + return p.failErr + } + p.mu.Lock() + defer p.mu.Unlock() + p.published = append(p.published, msg) + return nil +} + +func (p *fakeProducer) PublishBatch(_ context.Context, msgs []queue.Message) error { + if p.failErr != nil { + return p.failErr + } + p.mu.Lock() + defer p.mu.Unlock() + p.published = append(p.published, msgs...) + return nil +} + +func (p *fakeProducer) Close() error { return nil } + +func (p *fakeProducer) count() int { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.published) +} + +// fakeBuffer 는 in-memory JobBuffer mock. +type fakeBuffer struct { + mu sync.Mutex + enqueued map[string][][]byte + enqueueErr error +} + +func newFakeBuffer() *fakeBuffer { + return &fakeBuffer{enqueued: make(map[string][][]byte)} +} + +func (b *fakeBuffer) EnqueueJob(_ context.Context, label string, payload []byte, _ int64) error { + if b.enqueueErr != nil { + return b.enqueueErr + } + b.mu.Lock() + defer b.mu.Unlock() + b.enqueued[label] = append(b.enqueued[label], payload) + return nil +} + +func (b *fakeBuffer) DrainJobs(_ context.Context, label string, n int) ([][]byte, error) { + b.mu.Lock() + defer b.mu.Unlock() + items := b.enqueued[label] + if len(items) == 0 || n <= 0 { + return nil, nil + } + if n > len(items) { + n = len(items) + } + out := items[:n] + b.enqueued[label] = items[n:] + return out, nil +} + +func (b *fakeBuffer) JobBufferLen(_ context.Context, label string) (int64, error) { + b.mu.Lock() + defer b.mu.Unlock() + return int64(len(b.enqueued[label])), nil +} + +func (b *fakeBuffer) lenFor(label string) int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.enqueued[label]) +} + +func newTestLog() *logger.Logger { + return logger.New(logger.DefaultConfig()) +} + +// TestBufferingProducer_Publish_NormalRoutesToBuffer 는 normal topic 메시지가 buffer 로 라우팅되는지 검증. +func TestBufferingProducer_Publish_NormalRoutesToBuffer(t *testing.T) { + under := &fakeProducer{} + buf := newFakeBuffer() + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + err := p.Publish(context.Background(), queue.Message{ + Topic: queue.TopicCrawlNormal, + Key: []byte("k1"), + Value: []byte("v1"), + }) + require.NoError(t, err) + assert.Equal(t, 0, under.count(), "normal 은 underlying 직접 publish 안 함") + assert.Equal(t, 1, buf.lenFor("normal")) + assert.Equal(t, 0, buf.lenFor("low")) +} + +// TestBufferingProducer_Publish_LowRoutesToBuffer 는 low topic 메시지가 buffer 로 라우팅되는지 검증. +func TestBufferingProducer_Publish_LowRoutesToBuffer(t *testing.T) { + under := &fakeProducer{} + buf := newFakeBuffer() + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + err := p.Publish(context.Background(), queue.Message{ + Topic: queue.TopicCrawlLow, + Value: []byte("v"), + }) + require.NoError(t, err) + assert.Equal(t, 0, under.count()) + assert.Equal(t, 1, buf.lenFor("low")) +} + +// TestBufferingProducer_Publish_HighDirectPublish 는 high priority 가 buffer 우회하여 직접 publish 되는지 검증. +func TestBufferingProducer_Publish_HighDirectPublish(t *testing.T) { + under := &fakeProducer{} + buf := newFakeBuffer() + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + err := p.Publish(context.Background(), queue.Message{ + Topic: queue.TopicCrawlHigh, + Value: []byte("v"), + }) + require.NoError(t, err) + assert.Equal(t, 1, under.count(), "high 는 underlying 직접 publish") + assert.Equal(t, 0, buf.lenFor("normal")) + assert.Equal(t, 0, buf.lenFor("low")) +} + +// TestBufferingProducer_Publish_NonCrawlTopicDirectPublish 는 crawl 외 토픽은 우회하는지 검증. +func TestBufferingProducer_Publish_NonCrawlTopicDirectPublish(t *testing.T) { + under := &fakeProducer{} + buf := newFakeBuffer() + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + err := p.Publish(context.Background(), queue.Message{ + Topic: "issuetracker.normalized", // crawl 외 + Value: []byte("v"), + }) + require.NoError(t, err) + assert.Equal(t, 1, under.count()) +} + +// TestBufferingProducer_Publish_BufferFailFallback 은 buffer enqueue 실패 시 underlying 으로 fallback 검증. +func TestBufferingProducer_Publish_BufferFailFallback(t *testing.T) { + under := &fakeProducer{} + buf := newFakeBuffer() + buf.enqueueErr = errors.New("redis down") + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + err := p.Publish(context.Background(), queue.Message{ + Topic: queue.TopicCrawlNormal, + Value: []byte("v"), + }) + require.NoError(t, err, "fallback 으로 underlying.Publish 가 성공하면 본 호출도 성공") + assert.Equal(t, 1, under.count(), "buffer 실패 시 underlying 으로 fallback") +} + +// TestBufferingProducer_PublishBatch_MixedPriorities 는 batch 안 high/normal/low 혼합 라우팅 검증. +func TestBufferingProducer_PublishBatch_MixedPriorities(t *testing.T) { + under := &fakeProducer{} + buf := newFakeBuffer() + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + msgs := []queue.Message{ + {Topic: queue.TopicCrawlHigh, Value: []byte("h1")}, + {Topic: queue.TopicCrawlNormal, Value: []byte("n1")}, + {Topic: queue.TopicCrawlLow, Value: []byte("l1")}, + {Topic: queue.TopicCrawlHigh, Value: []byte("h2")}, + {Topic: queue.TopicCrawlNormal, Value: []byte("n2")}, + } + require.NoError(t, p.PublishBatch(context.Background(), msgs)) + + assert.Equal(t, 2, under.count(), "high 2개만 underlying 으로 직접") + assert.Equal(t, 2, buf.lenFor("normal")) + assert.Equal(t, 1, buf.lenFor("low")) +} + +// TestBufferingProducer_NilBufferUsesNoop 은 nil buffer 가 NoopJobBuffer 로 대체되어 fallback 됨을 검증. +func TestBufferingProducer_NilBufferUsesNoop(t *testing.T) { + under := &fakeProducer{} + p := queue.NewBufferingProducer(under, nil, 0, newTestLog()) // nil buffer + + err := p.Publish(context.Background(), queue.Message{ + Topic: queue.TopicCrawlNormal, + Value: []byte("v"), + }) + require.NoError(t, err) + assert.Equal(t, 1, under.count(), "nil buffer → Noop 가 항상 error → fallback 으로 underlying 호출") +} + +// TestBufferingProducer_Underlying 은 Underlying() 이 데코 우회용 producer 를 반환하는지 검증. +func TestBufferingProducer_Underlying(t *testing.T) { + under := &fakeProducer{} + buf := newFakeBuffer() + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + got := p.Underlying() + assert.Same(t, under, got, "Underlying 은 wrapping 전 raw producer 그대로 반환") +} + +// TestEncodeDecodeBufferedMessage 는 buffered 직렬화 round-trip 을 검증. +func TestEncodeDecodeBufferedMessage(t *testing.T) { + under := &fakeProducer{} + buf := newFakeBuffer() + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + orig := queue.Message{ + Topic: queue.TopicCrawlNormal, + Key: []byte("key-123"), + Value: []byte(`{"hello":"world"}`), + Headers: map[string]string{"crawler": "test", "priority": "1"}, + } + require.NoError(t, p.Publish(context.Background(), orig)) + + // buffer 에서 payload 를 꺼내 직접 decode. + payloads, err := buf.DrainJobs(context.Background(), "normal", 10) + require.NoError(t, err) + require.Len(t, payloads, 1) + + decoded, bufferedAt, err := queue.DecodeBufferedMessage(payloads[0]) + require.NoError(t, err) + assert.Equal(t, orig.Topic, decoded.Topic) + assert.Equal(t, orig.Key, decoded.Key) + assert.Equal(t, orig.Value, decoded.Value) + assert.Equal(t, orig.Headers, decoded.Headers) + assert.False(t, bufferedAt.IsZero(), "BufferedAt 은 enqueue 시점에 채워짐") +} From b0d5874085a5d1aa5a47eef0a6b2fc9614ae50c0 Mon Sep 17 00:00:00 2001 From: juhyuni Date: Mon, 18 May 2026 23:00:05 +0900 Subject: [PATCH 3/7] =?UTF-8?q?[FEAT]:=20BufferDrainer=20goroutine=20?= =?UTF-8?q?=E2=80=94=20Kafka=20backlog=20=EA=B8=B0=EB=B0=98=20=EC=A3=BC?= =?UTF-8?q?=EA=B8=B0=EC=A0=81=20drain=20(=EC=9D=B4=EC=8A=88=20#510)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - internal/scheduler/buffer_drainer.go: 별도 goroutine - 매 tick (default 30s) 마다 normal/low 각 priority 별로: - queue.BacklogChecker.Backlog 로 현재 lag 조회 - available = TargetBacklog - lag (음수면 skip) - n := min(available, DrainBatch, JobBuffer.JobBufferLen) - JobBuffer.DrainJobs → underlying.PublishBatch - publish 실패 시 drained payload 재적재 (best-effort, 순서 보존 X) - Start/Stop graceful shutdown 지원 — 진행 중 cycle 완료 대기 - 부팅 직후 1회 즉시 drain — 이전 세션 잔존물 회복 - 단위 테스트 5개 (mock buffer + producer + checker): drain available / skip on threshold / re-enqueue on fail / idle / nil deps validation Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/scheduler/buffer_drainer.go | 261 ++++++++++++++++ .../internal/scheduler/buffer_drainer_test.go | 286 ++++++++++++++++++ 2 files changed, 547 insertions(+) create mode 100644 internal/scheduler/buffer_drainer.go create mode 100644 test/internal/scheduler/buffer_drainer_test.go diff --git a/internal/scheduler/buffer_drainer.go b/internal/scheduler/buffer_drainer.go new file mode 100644 index 00000000..9034e4bf --- /dev/null +++ b/internal/scheduler/buffer_drainer.go @@ -0,0 +1,261 @@ +package scheduler + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "issuetracker/pkg/logger" + "issuetracker/pkg/queue" +) + +// BufferDrainer 는 Redis JobBuffer 에 적재된 normal/low priority crawl 메시지를 +// Kafka backlog 가 임계 미만일 때 점진적으로 underlying Producer 로 publish 합니다 (이슈 #510). +// +// 동작 (주기적 tick): +// 1. priority 별로 Kafka topic 의 현재 backlog (consumer-group lag) 를 BacklogChecker 로 조회 +// 2. available = targetBacklog - currentBacklog +// 3. n := min(available, drainBatch, JobBuffer.JobBufferLen) +// 4. JobBuffer.DrainJobs(label, n) → underlying.PublishBatch +// 5. publish 실패 시 drained payload 들을 다시 EnqueueJob 으로 재적재 (순서 보존 X — fail-safe) +// +// Stop 호출 시 다음 tick 전에 종료. 이미 drain 중인 사이클은 완료까지 대기 (graceful). +type BufferDrainer struct { + buffer queue.JobBuffer + producer queue.Producer // underlying — buffering 데코레이터 아닌 raw Kafka producer + checker queue.BacklogChecker + groupID string + targetBacklog int64 + drainBatch int + maxLen int64 // 재적재 시 LTRIM cap (BufferingProducer 와 동일 값) + interval time.Duration + checkTimeout time.Duration + log *logger.Logger + + wg sync.WaitGroup + stopCh chan struct{} + once sync.Once +} + +// BufferDrainerConfig 는 BufferDrainer 생성용 설정입니다. +type BufferDrainerConfig struct { + // Interval — 매 tick 의 주기. 권장 30s. + Interval time.Duration + // TargetBacklog — 유지하려는 Kafka lag 상한. drain 후 backlog 가 이 값 이하로 유지되도록. + // scheduler.BacklogThrottler 의 MaxBacklog (보통 5000) 의 60% 권장 (예: 3000). + TargetBacklog int64 + // DrainBatch — 한 tick 당 priority 별로 최대 drain 할 메시지 수. 권장 100. + DrainBatch int + // MaxLen — 재적재 시 buffer LIST 최대 길이 (BufferingProducer 와 동일 값 권장). + MaxLen int64 + // CheckTimeout — Backlog() 호출 한 번에 적용할 deadline. 0 이면 ctx 만 사용. + CheckTimeout time.Duration + // GroupID — Kafka consumer group ID (보통 queue.GroupCrawlerWorkers). + GroupID string +} + +// NewBufferDrainer 는 BufferDrainer 를 생성합니다. +// +// buffer / producer / checker / log 가 nil 이면 nil + error (silent crash 회피). +// cfg.Interval / DrainBatch / TargetBacklog 가 0 이하면 합리적 default 적용. +func NewBufferDrainer( + buffer queue.JobBuffer, + producer queue.Producer, + checker queue.BacklogChecker, + cfg BufferDrainerConfig, + log *logger.Logger, +) (*BufferDrainer, error) { + if buffer == nil { + return nil, errors.New("buffer drainer: nil JobBuffer") + } + if producer == nil { + return nil, errors.New("buffer drainer: nil underlying Producer") + } + if checker == nil { + return nil, errors.New("buffer drainer: nil BacklogChecker") + } + if log == nil { + return nil, errors.New("buffer drainer: nil logger") + } + if cfg.Interval <= 0 { + cfg.Interval = 30 * time.Second + } + if cfg.DrainBatch <= 0 { + cfg.DrainBatch = 100 + } + if cfg.TargetBacklog <= 0 { + cfg.TargetBacklog = 3000 + } + if cfg.GroupID == "" { + cfg.GroupID = queue.GroupCrawlerWorkers + } + return &BufferDrainer{ + buffer: buffer, + producer: producer, + checker: checker, + groupID: cfg.GroupID, + targetBacklog: cfg.TargetBacklog, + drainBatch: cfg.DrainBatch, + maxLen: cfg.MaxLen, + interval: cfg.Interval, + checkTimeout: cfg.CheckTimeout, + log: log, + stopCh: make(chan struct{}), + }, nil +} + +// drainTargets 는 buffer label → Kafka topic 매핑입니다. high 는 buffer 사용 안 함. +var drainTargets = []struct { + label string + topic string +}{ + {label: "normal", topic: queue.TopicCrawlNormal}, + {label: "low", topic: queue.TopicCrawlLow}, +} + +// Start 는 background goroutine 을 띄워 주기적 drain 을 시작합니다. +// ctx 는 long-lived parent — cancel 되면 다음 tick 에 종료. Stop() 도 동등 효과. +func (d *BufferDrainer) Start(ctx context.Context) { + d.wg.Add(1) + go d.run(ctx) + d.log.WithFields(map[string]interface{}{ + "interval": d.interval.String(), + "target_backlog": d.targetBacklog, + "drain_batch": d.drainBatch, + }).Info("buffer drainer started") +} + +// Stop 은 다음 tick 전에 drainer 를 정지합니다. 이미 진행 중인 drainOnce 는 완료까지 대기. +func (d *BufferDrainer) Stop() { + d.once.Do(func() { + close(d.stopCh) + }) + d.wg.Wait() + d.log.Info("buffer drainer stopped") +} + +func (d *BufferDrainer) run(ctx context.Context) { + defer d.wg.Done() + ticker := time.NewTicker(d.interval) + defer ticker.Stop() + + // 부팅 직후 1회 즉시 drain — buffer 에 이전 세션 잔존물이 있을 수 있음. + d.drainAll(ctx) + + for { + select { + case <-ctx.Done(): + return + case <-d.stopCh: + return + case <-ticker.C: + d.drainAll(ctx) + } + } +} + +func (d *BufferDrainer) drainAll(ctx context.Context) { + for _, tgt := range drainTargets { + if err := d.drainOnce(ctx, tgt.label, tgt.topic); err != nil { + // 개별 priority 실패가 다른 priority 의 drain 을 막지 않도록. + d.log.WithFields(map[string]interface{}{ + "label": tgt.label, + "topic": tgt.topic, + }).WithError(err).Warn("buffer drain cycle failed (non-fatal)") + } + } +} + +func (d *BufferDrainer) drainOnce(ctx context.Context, label, topic string) error { + // 1) 현재 backlog 조회 + checkCtx := ctx + if d.checkTimeout > 0 { + var cancel context.CancelFunc + checkCtx, cancel = context.WithTimeout(ctx, d.checkTimeout) + defer cancel() + } + backlog, err := d.checker.Backlog(checkCtx, topic, d.groupID) + if err != nil { + // backlog 조회 실패 → fail-closed (이번 cycle drain skip). throttle 과 반대 — 여기서 + // fail-open 하면 backlog 추정 없이 무제한 drain → Kafka 과부하 위험. + return fmt.Errorf("backlog check for %s: %w", topic, err) + } + + available := d.targetBacklog - backlog + if available <= 0 { + d.log.WithFields(map[string]interface{}{ + "label": label, + "topic": topic, + "backlog": backlog, + "target_backlog": d.targetBacklog, + }).Debug("buffer drain skipped — target backlog reached") + return nil + } + + // 2) buffer 현재 길이 조회 — drain 할 양 결정 + bufLen, err := d.buffer.JobBufferLen(ctx, label) + if err != nil { + return fmt.Errorf("buffer len %s: %w", label, err) + } + if bufLen == 0 { + return nil // idle — 정상 + } + + n := int(available) + if d.drainBatch < n { + n = d.drainBatch + } + if int(bufLen) < n { + n = int(bufLen) + } + + // 3) drain + payloads, err := d.buffer.DrainJobs(ctx, label, n) + if err != nil { + return fmt.Errorf("drain %s: %w", label, err) + } + if len(payloads) == 0 { + return nil + } + + // 4) underlying Producer 로 publish + msgs := make([]queue.Message, 0, len(payloads)) + for _, p := range payloads { + msg, _, decErr := queue.DecodeBufferedMessage(p) + if decErr != nil { + d.log.WithError(decErr).Warn("buffer drain decode failed, dropping payload") + continue + } + msgs = append(msgs, msg) + } + if len(msgs) == 0 { + return nil + } + + if pubErr := d.producer.PublishBatch(ctx, msgs); pubErr != nil { + // publish 실패 → 재적재 (순서 보존 X). MaxLen 으로 LTRIM 회피 위해 maxLen 그대로 전달. + d.log.WithFields(map[string]interface{}{ + "label": label, + "topic": topic, + "count": len(msgs), + }).WithError(pubErr).Warn("buffer drain publish failed, re-enqueueing payloads") + for _, p := range payloads { + if reErr := d.buffer.EnqueueJob(ctx, label, p, d.maxLen); reErr != nil { + d.log.WithError(reErr).Warn("re-enqueue after publish failure failed (data loss possible)") + } + } + return fmt.Errorf("publish batch %s: %w", topic, pubErr) + } + + d.log.WithFields(map[string]interface{}{ + "label": label, + "topic": topic, + "drained": len(msgs), + "backlog_before": backlog, + "target_backlog": d.targetBacklog, + "buffer_len_left": bufLen - int64(len(payloads)), + }).Info("buffer drained to kafka") + return nil +} diff --git a/test/internal/scheduler/buffer_drainer_test.go b/test/internal/scheduler/buffer_drainer_test.go new file mode 100644 index 00000000..77e35a6b --- /dev/null +++ b/test/internal/scheduler/buffer_drainer_test.go @@ -0,0 +1,286 @@ +package scheduler_test + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "issuetracker/internal/scheduler" + "issuetracker/pkg/logger" + "issuetracker/pkg/queue" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Mocks (BufferDrainer) +// ───────────────────────────────────────────────────────────────────────────── + +type mockJobBuffer struct { + mu sync.Mutex + store map[string][][]byte + enqErr error + drainErr error + lenErr error +} + +func newMockJobBuffer() *mockJobBuffer { + return &mockJobBuffer{store: make(map[string][][]byte)} +} + +func (b *mockJobBuffer) EnqueueJob(_ context.Context, label string, payload []byte, _ int64) error { + if b.enqErr != nil { + return b.enqErr + } + b.mu.Lock() + defer b.mu.Unlock() + b.store[label] = append(b.store[label], payload) + return nil +} + +func (b *mockJobBuffer) DrainJobs(_ context.Context, label string, n int) ([][]byte, error) { + if b.drainErr != nil { + return nil, b.drainErr + } + b.mu.Lock() + defer b.mu.Unlock() + items := b.store[label] + if n <= 0 || len(items) == 0 { + return nil, nil + } + if n > len(items) { + n = len(items) + } + out := append([][]byte(nil), items[:n]...) + b.store[label] = items[n:] + return out, nil +} + +func (b *mockJobBuffer) JobBufferLen(_ context.Context, label string) (int64, error) { + if b.lenErr != nil { + return 0, b.lenErr + } + b.mu.Lock() + defer b.mu.Unlock() + return int64(len(b.store[label])), nil +} + +func (b *mockJobBuffer) seed(label string, payloads ...[]byte) { + b.mu.Lock() + defer b.mu.Unlock() + b.store[label] = append(b.store[label], payloads...) +} + +type mockDrainerProducer struct { + mu sync.Mutex + published []queue.Message + failErr error +} + +func (p *mockDrainerProducer) Publish(_ context.Context, msg queue.Message) error { + if p.failErr != nil { + return p.failErr + } + p.mu.Lock() + defer p.mu.Unlock() + p.published = append(p.published, msg) + return nil +} + +func (p *mockDrainerProducer) PublishBatch(_ context.Context, msgs []queue.Message) error { + if p.failErr != nil { + return p.failErr + } + p.mu.Lock() + defer p.mu.Unlock() + p.published = append(p.published, msgs...) + return nil +} + +func (p *mockDrainerProducer) Close() error { return nil } + +func (p *mockDrainerProducer) count() int { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.published) +} + +type fixedBacklogChecker struct { + lag atomic.Int64 + err error +} + +func (c *fixedBacklogChecker) Backlog(_ context.Context, _ string, _ string) (int64, error) { + if c.err != nil { + return 0, c.err + } + return c.lag.Load(), nil +} + +func newDrainerTestLog() *logger.Logger { + return logger.New(logger.DefaultConfig()) +} + +// encodeMsg 는 BufferDrainer 가 decode 가능한 직렬화 payload 를 만들어 줍니다. +// 본 헬퍼는 BufferingProducer 의 encodeBufferedMessage 와 동일 포맷이어야 합니다. +// queue 패키지의 unexported encodeBufferedMessage 직접 호출이 불가능하므로, +// BufferingProducer.Publish 를 통해 mock buffer 에 적재 → seed payload 로 사용합니다. +func encodeMsg(t *testing.T, msg queue.Message) []byte { + t.Helper() + relay := &mockJobBuffer{store: make(map[string][][]byte)} + bp := queue.NewBufferingProducer(&mockDrainerProducer{}, relay, 0, newDrainerTestLog()) + require.NoError(t, bp.Publish(context.Background(), msg)) + label := "normal" + if msg.Topic == queue.TopicCrawlLow { + label = "low" + } + out, err := relay.DrainJobs(context.Background(), label, 1) + require.NoError(t, err) + require.Len(t, out, 1) + return out[0] +} + +// ───────────────────────────────────────────────────────────────────────────── +// Tests +// ───────────────────────────────────────────────────────────────────────────── + +// TestBufferDrainer_DrainsUpToAvailable 은 backlog < target 시 buffer 에서 정확한 수만큼 drain 후 publish 검증. +func TestBufferDrainer_DrainsUpToAvailable(t *testing.T) { + buf := newMockJobBuffer() + for i := 0; i < 5; i++ { + payload := encodeMsg(t, queue.Message{Topic: queue.TopicCrawlNormal, Value: []byte("v")}) + buf.seed("normal", payload) + } + + prod := &mockDrainerProducer{} + check := &fixedBacklogChecker{} + check.lag.Store(100) // 현재 lag + + d, err := scheduler.NewBufferDrainer(buf, prod, check, scheduler.BufferDrainerConfig{ + Interval: time.Hour, // tick 발화 회피 — drainAll 한 번만 검증 + TargetBacklog: 200, // available = 100 + DrainBatch: 50, // → min(100, 50, 5) = 5 + GroupID: queue.GroupCrawlerWorkers, + }, newDrainerTestLog()) + require.NoError(t, err) + + // drainAll 은 unexported — Start + ctx cancel 로 간접 호출. + ctx, cancel := context.WithCancel(context.Background()) + d.Start(ctx) + + // drainAll 이 부팅 직후 1회 호출 — 약간 대기 후 cancel. + time.Sleep(80 * time.Millisecond) + cancel() + d.Stop() + + assert.Equal(t, 5, prod.count(), "buffer 5건 모두 drain 후 underlying publish") +} + +// TestBufferDrainer_SkipsWhenBacklogReached 은 backlog >= target 시 drain skip 검증. +func TestBufferDrainer_SkipsWhenBacklogReached(t *testing.T) { + buf := newMockJobBuffer() + for i := 0; i < 3; i++ { + payload := encodeMsg(t, queue.Message{Topic: queue.TopicCrawlNormal, Value: []byte("v")}) + buf.seed("normal", payload) + } + + prod := &mockDrainerProducer{} + check := &fixedBacklogChecker{} + check.lag.Store(5000) // 매우 큰 lag + + d, err := scheduler.NewBufferDrainer(buf, prod, check, scheduler.BufferDrainerConfig{ + Interval: time.Hour, + TargetBacklog: 3000, // available = -2000 → skip + DrainBatch: 100, + GroupID: queue.GroupCrawlerWorkers, + }, newDrainerTestLog()) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + d.Start(ctx) + time.Sleep(80 * time.Millisecond) + cancel() + d.Stop() + + assert.Equal(t, 0, prod.count(), "backlog 임계 도달 — drain 0") + // buffer 에 그대로 남아있어야 함 + n, _ := buf.JobBufferLen(context.Background(), "normal") + assert.Equal(t, int64(3), n) +} + +// TestBufferDrainer_ReEnqueueOnPublishFailure 는 publish 실패 시 drained payload 가 재적재되는지 검증. +func TestBufferDrainer_ReEnqueueOnPublishFailure(t *testing.T) { + buf := newMockJobBuffer() + for i := 0; i < 3; i++ { + payload := encodeMsg(t, queue.Message{Topic: queue.TopicCrawlNormal, Value: []byte("v")}) + buf.seed("normal", payload) + } + + prod := &mockDrainerProducer{failErr: errors.New("kafka down")} + check := &fixedBacklogChecker{} + check.lag.Store(0) + + d, err := scheduler.NewBufferDrainer(buf, prod, check, scheduler.BufferDrainerConfig{ + Interval: time.Hour, + TargetBacklog: 1000, + DrainBatch: 100, + GroupID: queue.GroupCrawlerWorkers, + }, newDrainerTestLog()) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + d.Start(ctx) + time.Sleep(80 * time.Millisecond) + cancel() + d.Stop() + + assert.Equal(t, 0, prod.count(), "publish 실패 → underlying 에 아무것도 안 들어감") + n, _ := buf.JobBufferLen(context.Background(), "normal") + assert.Equal(t, int64(3), n, "실패한 drained payload 가 buffer 에 재적재되어 잔존") +} + +// TestBufferDrainer_IdleWhenBufferEmpty 는 buffer 비어있을 때 publish 호출 없음 검증. +func TestBufferDrainer_IdleWhenBufferEmpty(t *testing.T) { + buf := newMockJobBuffer() + prod := &mockDrainerProducer{} + check := &fixedBacklogChecker{} + check.lag.Store(0) + + d, err := scheduler.NewBufferDrainer(buf, prod, check, scheduler.BufferDrainerConfig{ + Interval: time.Hour, + TargetBacklog: 1000, + DrainBatch: 100, + GroupID: queue.GroupCrawlerWorkers, + }, newDrainerTestLog()) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + d.Start(ctx) + time.Sleep(80 * time.Millisecond) + cancel() + d.Stop() + + assert.Equal(t, 0, prod.count()) +} + +// TestBufferDrainer_NilDepsReturnError 는 필수 의존성 nil 시 생성자가 error 반환을 검증. +func TestBufferDrainer_NilDepsReturnError(t *testing.T) { + buf := newMockJobBuffer() + prod := &mockDrainerProducer{} + check := &fixedBacklogChecker{} + log := newDrainerTestLog() + cfg := scheduler.BufferDrainerConfig{Interval: time.Second, TargetBacklog: 1000} + + _, err := scheduler.NewBufferDrainer(nil, prod, check, cfg, log) + assert.Error(t, err, "nil buffer") + _, err = scheduler.NewBufferDrainer(buf, nil, check, cfg, log) + assert.Error(t, err, "nil producer") + _, err = scheduler.NewBufferDrainer(buf, prod, nil, cfg, log) + assert.Error(t, err, "nil checker") + _, err = scheduler.NewBufferDrainer(buf, prod, check, cfg, nil) + assert.Error(t, err, "nil logger") +} From 5a2abdef1d46cb575fca401625140322593be398 Mon Sep 17 00:00:00 2001 From: juhyuni Date: Mon, 18 May 2026 23:00:12 +0900 Subject: [PATCH 4/7] =?UTF-8?q?[FEAT]:=20JobBufferConfig=206=EA=B0=9C=20en?= =?UTF-8?q?v=20+=20PositiveInt64=20parse=20helper=20(=EC=9D=B4=EC=8A=88=20?= =?UTF-8?q?#510)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pkg/config/processor/job_buffer.go: PUBLISHER_REDIS_BUFFER_* - ENABLED (default false — opt-in, backward compatible) - DRAIN_INTERVAL (30s) / TARGET_BACKLOG (3000) / DRAIN_BATCH (100) - MAX_LEN (100000 — LTRIM cap) / CHECK_TIMEOUT (5s) - pkg/config/internal/parse: PositiveInt64 helper 추가 (TargetBacklog 검증용) - .env.example: 신규 env 6개 + 운영 가이드 주석 Co-Authored-By: Claude Opus 4.7 (1M context) --- .env.example | 12 +++++ pkg/config/internal/parse/parse.go | 17 +++++++ pkg/config/processor/job_buffer.go | 82 ++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+) create mode 100644 pkg/config/processor/job_buffer.go diff --git a/.env.example b/.env.example index 332d0e15..f615cc65 100644 --- a/.env.example +++ b/.env.example @@ -93,6 +93,18 @@ SCHEDULER_MAX_RETRIES=3 SCHEDULER_MAX_BACKLOG=5000 SCHEDULER_BACKLOG_CHECK_TIMEOUT=5s +# Publisher Redis 버퍼 (normal/low priority crawl 토픽 임시 적재, 이슈 #510) +# - Enabled=true 시 BufferingProducer + BufferDrainer goroutine wiring +# - drainer 가 매 DRAIN_INTERVAL 마다 Kafka consumer-group lag 확인 후 TARGET_BACKLOG 이하 유지하도록 Redis → Kafka publish +# - high priority 는 직접 publish (영향 X) +# - Redis 미연결 시 자동 비활성 (graceful degrade) +PUBLISHER_REDIS_BUFFER_ENABLED=false +PUBLISHER_REDIS_BUFFER_DRAIN_INTERVAL=30s +PUBLISHER_REDIS_BUFFER_TARGET_BACKLOG=3000 +PUBLISHER_REDIS_BUFFER_DRAIN_BATCH=100 +PUBLISHER_REDIS_BUFFER_MAX_LEN=100000 +PUBLISHER_REDIS_BUFFER_CHECK_TIMEOUT=5s + # Classifier 서비스 연결 설정 CLASSIFIER_HTTP_ADDR=http://localhost:8000 CLASSIFIER_GRPC_ADDR=localhost:50051 diff --git a/pkg/config/internal/parse/parse.go b/pkg/config/internal/parse/parse.go index d2838b41..b568c370 100644 --- a/pkg/config/internal/parse/parse.go +++ b/pkg/config/internal/parse/parse.go @@ -148,6 +148,23 @@ func NonNegativeInt32(key string, dest *int32) error { return nil } +// PositiveInt64 는 PositiveInt 의 int64 변형 (> 0). +func PositiveInt64(key string, dest *int64) error { + v := os.Getenv(key) + if v == "" { + return nil + } + n, err := strconv.ParseInt(v, 10, 64) + if err != nil { + return fmt.Errorf("parse %s %q: %w", key, v, err) + } + if n <= 0 { + return fmt.Errorf("%s %d must be positive", key, n) + } + *dest = n + return nil +} + // NonNegativeInt64 는 NonNegativeInt 의 int64 변형. func NonNegativeInt64(key string, dest *int64) error { v := os.Getenv(key) diff --git a/pkg/config/processor/job_buffer.go b/pkg/config/processor/job_buffer.go new file mode 100644 index 00000000..739bde84 --- /dev/null +++ b/pkg/config/processor/job_buffer.go @@ -0,0 +1,82 @@ +package processorcfg + +import ( + "errors" + "fmt" + "os" + "time" + + "github.com/joho/godotenv" + + "issuetracker/pkg/config/internal/parse" +) + +// JobBufferConfig 는 normal/low priority crawl 메시지의 Redis 버퍼링 정책 설정입니다 (이슈 #510). +// +// Enabled=false (default) 면 BufferingProducer wiring 자체 skip — 기존 직접 publish 동작 유지. +// Enabled=true 면 cmd/issuetracker/main.go 가 BufferingProducer + BufferDrainer 를 wire. +type JobBufferConfig struct { + // Enabled — 본 기능 활성화. PUBLISHER_REDIS_BUFFER_ENABLED (default: false). + Enabled bool + + // DrainInterval — BufferDrainer goroutine 의 tick 주기. + // PUBLISHER_REDIS_BUFFER_DRAIN_INTERVAL (default: 30s). + DrainInterval time.Duration + + // TargetBacklog — drainer 가 유지하려는 Kafka consumer-group lag 상한. + // scheduler.MaxBacklog (보통 5000) 의 60% 권장 (예: 3000). + // PUBLISHER_REDIS_BUFFER_TARGET_BACKLOG (default: 3000). + TargetBacklog int64 + + // DrainBatch — 한 tick 당 priority 별로 최대 drain 할 메시지 수. + // PUBLISHER_REDIS_BUFFER_DRAIN_BATCH (default: 100). + DrainBatch int + + // MaxLen — Redis LIST 의 최대 길이. >0 이면 EnqueueJob 시 LTRIM 으로 oldest 제거. + // 0 이면 길이 제한 없음 (운영자가 모니터링 책임). + // PUBLISHER_REDIS_BUFFER_MAX_LEN (default: 100000). + MaxLen int64 + + // CheckTimeout — Backlog() 호출 한 번에 적용할 deadline. + // PUBLISHER_REDIS_BUFFER_CHECK_TIMEOUT (default: 5s). + CheckTimeout time.Duration +} + +// DefaultJobBufferConfig 는 기본 JobBufferConfig 를 반환합니다. +func DefaultJobBufferConfig() JobBufferConfig { + return JobBufferConfig{ + Enabled: false, + DrainInterval: 30 * time.Second, + TargetBacklog: 3000, + DrainBatch: 100, + MaxLen: 100000, + CheckTimeout: 5 * time.Second, + } +} + +// LoadJobBuffer 는 .env + 환경변수로 JobBufferConfig 를 구성합니다. +func LoadJobBuffer(envFiles ...string) (JobBufferConfig, error) { + if len(envFiles) == 0 { + envFiles = []string{".env"} + } + if err := godotenv.Load(envFiles...); err != nil && !errors.Is(err, os.ErrNotExist) { + return JobBufferConfig{}, fmt.Errorf("failed to load env files %v: %w", envFiles, err) + } + + cfg := DefaultJobBufferConfig() + + for _, op := range []error{ + parse.Bool("PUBLISHER_REDIS_BUFFER_ENABLED", &cfg.Enabled), + parse.PositiveDuration("PUBLISHER_REDIS_BUFFER_DRAIN_INTERVAL", &cfg.DrainInterval), + parse.PositiveInt64("PUBLISHER_REDIS_BUFFER_TARGET_BACKLOG", &cfg.TargetBacklog), + parse.PositiveInt("PUBLISHER_REDIS_BUFFER_DRAIN_BATCH", &cfg.DrainBatch), + parse.NonNegativeInt64("PUBLISHER_REDIS_BUFFER_MAX_LEN", &cfg.MaxLen), // 0 = unlimited + parse.PositiveDuration("PUBLISHER_REDIS_BUFFER_CHECK_TIMEOUT", &cfg.CheckTimeout), + } { + if op != nil { + return JobBufferConfig{}, op + } + } + + return cfg, nil +} From 21d200c5a5eba48f87861d62f1f8196411e6cf0e Mon Sep 17 00:00:00 2001 From: juhyuni Date: Mon, 18 May 2026 23:00:30 +0900 Subject: [PATCH 5/7] =?UTF-8?q?[FEAT]:=20cmd/issuetracker=20BufferingProdu?= =?UTF-8?q?cer=20+=20BufferDrainer=20wiring=20(=EC=9D=B4=EC=8A=88=20#510)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Redis 클라이언트 초기화를 crawlerProducer 생성 전으로 이동 — JobBuffer / ProcessingLock / IngestionLock / RetryScheduler 가 공유 - LoadJobBuffer + bufferCfg.Enabled && redisClient != nil 시 wiring: - rawCrawlerProducer (KafkaProducer) 를 BufferingProducer 로 감싸 jobPublisher 에 주입 - BufferDrainer 는 rawCrawlerProducer (underlying) 로 직접 publish — 무한 루프 회피 - 활성화 로그에 drain_interval / target_backlog / drain_batch / max_len 출력 - 기능 비활성 (default) 시 raw producer 직접 사용 — 기존 동작 100% 보존 - Redis 미연결 시 자동 비활성 + WARN — graceful degrade - shutdown 체인: sched.Stop() 직후 bufferDrainer.Stop() — scheduler 의 마지막 publish 가 buffer 에 들어간 뒤 drainer 가 비울 기회 확보 Co-Authored-By: Claude Opus 4.7 (1M context) --- cmd/issuetracker/main.go | 157 +++++++++++++++++++++++++++------------ 1 file changed, 109 insertions(+), 48 deletions(-) diff --git a/cmd/issuetracker/main.go b/cmd/issuetracker/main.go index 23f03515..0fb67db5 100644 --- a/cmd/issuetracker/main.go +++ b/cmd/issuetracker/main.go @@ -130,8 +130,71 @@ func main() { crawlerKafkaCfg := queue.DefaultConfig() crawlerKafkaCfg.GroupID = queue.GroupCrawlerWorkers - crawlerProducer := queue.NewProducer(crawlerKafkaCfg) - defer crawlerProducer.Close() + // Redis client 조기 초기화 — JobBuffer / ProcessingLock / IngestionLock / RetryScheduler 가 + // 모두 공유 (이슈 #510). 실패 시 graceful degrade — 모든 Redis 기반 기능은 noop fallback. + var redisClientShared *redis.Client + redisCfg, err := storagecfg.LoadRedis() + if err != nil { + log.WithError(err).Warn("failed to load redis config, falling back to noop processing/ingestion lock and disabled job buffer") + } else { + redisClient, redisErr := redis.New(ctx, redisCfg) + if redisErr != nil { + log.WithError(redisErr).Warn("failed to connect to redis, falling back to noop processing/ingestion lock and disabled job buffer") + } else { + defer redisClient.Close() + redisClientShared = redisClient + log.WithFields(map[string]interface{}{ + "host": redisCfg.Host, + "port": redisCfg.Port, + }).Info("redis connected (shared by lock / ingestion lock / job buffer / retry scheduler)") + } + } + + // 이슈 #510 — normal/low priority crawl 토픽 Redis 버퍼링 (opt-in). + // Enabled + Redis 연결 시 BufferingProducer 데코레이터 + BufferDrainer goroutine 활성화. + // 비활성 시 raw KafkaProducer 사용 (기존 동작 보존). + jobBufferCfg, err := processorcfg.LoadJobBuffer() + if err != nil { + log.WithError(err).Fatal("failed to load job buffer config") + } + + rawCrawlerProducer := queue.NewProducer(crawlerKafkaCfg) + defer rawCrawlerProducer.Close() + + var crawlerProducer queue.Producer = rawCrawlerProducer + var bufferDrainer *scheduler.BufferDrainer // nil 이면 기능 비활성 — Stop() skip + + if jobBufferCfg.Enabled && redisClientShared != nil { + bufferingProducer := queue.NewBufferingProducer(rawCrawlerProducer, redisClientShared, jobBufferCfg.MaxLen, log) + crawlerProducer = bufferingProducer + + drainer, derr := scheduler.NewBufferDrainer( + redisClientShared, + rawCrawlerProducer, // drainer 는 underlying 으로 직접 publish — 무한 루프 회피 + queue.NewBacklogChecker(crawlerKafkaCfg.Brokers, jobBufferCfg.CheckTimeout), + scheduler.BufferDrainerConfig{ + Interval: jobBufferCfg.DrainInterval, + TargetBacklog: jobBufferCfg.TargetBacklog, + DrainBatch: jobBufferCfg.DrainBatch, + MaxLen: jobBufferCfg.MaxLen, + CheckTimeout: jobBufferCfg.CheckTimeout, + GroupID: queue.GroupCrawlerWorkers, + }, + log, + ) + if derr != nil { + log.WithError(derr).Fatal("failed to construct buffer drainer") + } + bufferDrainer = drainer + log.WithFields(map[string]interface{}{ + "drain_interval": jobBufferCfg.DrainInterval.String(), + "target_backlog": jobBufferCfg.TargetBacklog, + "drain_batch": jobBufferCfg.DrainBatch, + "max_len": jobBufferCfg.MaxLen, + }).Info("publisher redis buffer enabled (normal/low priority)") + } else if jobBufferCfg.Enabled && redisClientShared == nil { + log.Warn("publisher redis buffer requested but redis unavailable — falling back to direct kafka publish") + } // 이슈 #391 — PriorityResolver chain 이 publisher 측으로 이동 + 모든 PublishX 가 // resolver 통과 (메타 #385 Sub 6). ExplicitPriorityResolver 를 chain 1순위 로 등록 — @@ -319,57 +382,41 @@ func main() { } } - // Redis 기반 ProcessingLock: 동일 URL 이 여러 worker/인스턴스에서 단계별 (fetcher/parser/validator) - // 중복 처리되는 것을 방지합니다. 단일 인스턴스를 fetcher / parser / validator 가 공유 — - // 단계 구분은 ProcessingKey(stage, url) 의 stage prefix 로 처리. - // worker/manager 가 nil 을 NoopProcessingLock 로 fallback 처리하는 설계와 일관되게, - // Redis 초기화 실패 시에도 크롤링이 중단되지 않도록 graceful degrade 합니다. + // Redis 기반 ProcessingLock / IngestionLock / DelayedRetryScheduler — redisClientShared 가 + // 위쪽에서 이미 초기화됨 (이슈 #510 — JobBuffer 와 client 공유). 본 블록은 lock/retry 만 wire. + // 단일 인스턴스를 fetcher / parser / validator 가 공유 — 단계 구분은 ProcessingKey(stage, url) + // 의 stage prefix 로 처리. worker/manager 가 nil 을 NoopProcessingLock 로 fallback 처리. var procLock locks.ProcessingLock var ingestionLock locks.IngestionLock var retryScheduler bus.RetryScheduler var retrySchedulerStop func() - var redisClientShared *redis.Client // failure counter wiring 에서 재사용 - redisCfg, err := storagecfg.LoadRedis() - if err != nil { - log.WithError(err).Warn("failed to load redis config, falling back to noop processing lock and ingestion lock") - } else { - redisClient, redisErr := redis.New(ctx, redisCfg) - if redisErr != nil { - log.WithError(redisErr).Warn("failed to connect to redis, falling back to noop processing lock and ingestion lock") - } else { - defer redisClient.Close() - redisClientShared = redisClient - log.WithFields(map[string]interface{}{ - "host": redisCfg.Host, - "port": redisCfg.Port, - }).Info("redis connected for processing lock and ingestion lock") - procLock = locks.NewRedisProcessingLock(redisClient, locks.DefaultProcessingLockTTL) - ingestionLock = locks.NewRedisIngestionLock(redisClient, redisCfg.IngestionLockTTL) - - // Delayed retry queue: retry 를 Redis ZSET 에 보관하고 별도 - // goroutine 이 ScheduledAt 도달 시 Kafka 에 발행 — worker 슬롯 점유 회피. - // Redis 부재 시 worker 가 lazy 로 KafkaImmediateRetryScheduler 를 사용 (기존 동작). - retryCfg := bus.DefaultRedisRetrySchedulerConfig() - // idle heartbeat 압축 (이슈 #370) — pkg/config 로 env 로드 일관성 유지. - retrySchedCfg, retrySchedErr := runtimecfg.LoadRetryScheduler() - if retrySchedErr != nil { - log.WithError(retrySchedErr).Fatal("RETRY_HEARTBEAT_EVERY_N_IDLE_TICKS 로드 실패") - } - retryCfg.HeartbeatEveryNIdleTicks = retrySchedCfg.HeartbeatEveryNIdleTicks - redisRetry := bus.NewRedisDelayedRetryScheduler( - redisClient, jobPublisher, - retryCfg, - log, - ) - runCtx, cancelRun := context.WithCancel(ctx) - redisRetry.Start(runCtx) // 내부에서 wg.Add 후 go Run — 패닉 안전 - retryScheduler = redisRetry - retrySchedulerStop = func() { - cancelRun() - redisRetry.Stop() - } - log.Info("redis delayed retry queue enabled (worker slot occupancy on retry resolved)") + if redisClientShared != nil { + procLock = locks.NewRedisProcessingLock(redisClientShared, locks.DefaultProcessingLockTTL) + ingestionLock = locks.NewRedisIngestionLock(redisClientShared, redisCfg.IngestionLockTTL) + + // Delayed retry queue: retry 를 Redis ZSET 에 보관하고 별도 + // goroutine 이 ScheduledAt 도달 시 Kafka 에 발행 — worker 슬롯 점유 회피. + // Redis 부재 시 worker 가 lazy 로 KafkaImmediateRetryScheduler 를 사용 (기존 동작). + retryCfg := bus.DefaultRedisRetrySchedulerConfig() + // idle heartbeat 압축 (이슈 #370) — pkg/config 로 env 로드 일관성 유지. + retrySchedCfg, retrySchedErr := runtimecfg.LoadRetryScheduler() + if retrySchedErr != nil { + log.WithError(retrySchedErr).Fatal("RETRY_HEARTBEAT_EVERY_N_IDLE_TICKS 로드 실패") + } + retryCfg.HeartbeatEveryNIdleTicks = retrySchedCfg.HeartbeatEveryNIdleTicks + redisRetry := bus.NewRedisDelayedRetryScheduler( + redisClientShared, jobPublisher, + retryCfg, + log, + ) + runCtx, cancelRun := context.WithCancel(ctx) + redisRetry.Start(runCtx) // 내부에서 wg.Add 후 go Run — 패닉 안전 + retryScheduler = redisRetry + retrySchedulerStop = func() { + cancelRun() + redisRetry.Stop() } + log.Info("redis delayed retry queue enabled (worker slot occupancy on retry resolved)") } if retrySchedulerStop != nil { defer retrySchedulerStop() @@ -845,6 +892,12 @@ func main() { log.Info("scheduler disabled by STAGES_SCHEDULER_ENABLED=false") } + // 이슈 #510 — BufferDrainer 시작 (config Enabled + Redis 연결 시에만 wiring 됨). + // drainer 가 scheduler stage 와 독립 — fetcher 단계가 활성화된 어떤 인스턴스에서도 drain 가능. + if bufferDrainer != nil { + bufferDrainer.Start(ctx) + } + // ══════════════════════════════════════════════════════════════════════════ // Processor (Validate) // ══════════════════════════════════════════════════════════════════════════ @@ -1091,6 +1144,14 @@ func main() { sched.Stop() + // 이슈 #510 — BufferDrainer 정지. scheduler.Stop 이후 호출 — scheduler 가 BufferingProducer 로 + // 보내는 publish 가 끝난 뒤 drainer 가 잔존 buffer 를 마지막으로 비울 수 있도록. + // 단, drainer 는 자기 tick 주기로 동작 — Stop 은 다음 tick 직전에 cancel + 진행 중 cycle 완료 대기. + // 잔존 buffer 는 Redis 에 보존되어 다음 부팅 시 자동 회복 (옵션 A — 본 이슈 본문 참조). + if bufferDrainer != nil { + bufferDrainer.Stop() + } + // Stage 들을 정의 순서대로 stop — fetcher → parser → validate 순서로 정리. // 데이터 파이프라인 source-first 원칙: fetcher 가 먼저 Kafka 발행을 멈추면 downstream // stage 가 in-flight 메시지를 graceful drain 가능. parser.Stage 내부에서 lifecycle From 7a54f1231e28f3c665696075f1f9ff3786db7605 Mon Sep 17 00:00:00 2001 From: juhyuni Date: Mon, 18 May 2026 23:10:27 +0900 Subject: [PATCH 6/7] =?UTF-8?q?[FIX]:=20=ED=94=BC=EB=93=9C=EB=B0=B1=20?= =?UTF-8?q?=EB=B0=98=EC=98=81,=20JobBuffer.EnqueueBatch=20=EC=B6=94?= =?UTF-8?q?=EA=B0=80=20=E2=80=94=20Redis=20pipelining=20=EC=9C=BC=EB=A1=9C?= =?UTF-8?q?=20N=E2=86=921=20RTT=20(gemini)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - JobBuffer interface 에 EnqueueBatch(label, payloads, maxLen) 추가 - pkg/redis/Client.EnqueueBatch: LPush variadic args + optional LTRIM 을 단일 pipeline 으로 전송 - EnqueueJob 을 EnqueueBatch 로 delegate 리팩터링 — single source of truth - NoopJobBuffer.EnqueueBatch 도 동등 error fallback - BufferingProducer.PublishBatch: label 별 그룹핑 후 1회 EnqueueBatch — 같은 label 의 N개 메시지 1 RTT - BufferDrainer publish 실패 re-enqueue 경로도 EnqueueBatch 사용 — 실패 복구 시 round-trip 절감 - 단위 테스트 추가: Redis 통합 4개 (batch / empty / LTRIM / invalid payload) + BufferingProducer label-별 단일 호출 검증 1개 Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/scheduler/buffer_drainer.go | 12 ++-- pkg/queue/buffering_producer.go | 44 +++++++++--- pkg/redis/job_buffer.go | 50 ++++++++++---- .../internal/scheduler/buffer_drainer_test.go | 10 +++ test/pkg/queue/buffering_producer_test.go | 53 +++++++++++++++ test/pkg/redis/job_buffer_test.go | 67 +++++++++++++++++++ 6 files changed, 209 insertions(+), 27 deletions(-) diff --git a/internal/scheduler/buffer_drainer.go b/internal/scheduler/buffer_drainer.go index 9034e4bf..82b8afdc 100644 --- a/internal/scheduler/buffer_drainer.go +++ b/internal/scheduler/buffer_drainer.go @@ -235,16 +235,18 @@ func (d *BufferDrainer) drainOnce(ctx context.Context, label, topic string) erro } if pubErr := d.producer.PublishBatch(ctx, msgs); pubErr != nil { - // publish 실패 → 재적재 (순서 보존 X). MaxLen 으로 LTRIM 회피 위해 maxLen 그대로 전달. + // publish 실패 → 재적재 (순서 보존 X). EnqueueBatch 로 1 RTT — N개 EnqueueJob 회피 + // (gemini PR #511 피드백). d.log.WithFields(map[string]interface{}{ "label": label, "topic": topic, "count": len(msgs), }).WithError(pubErr).Warn("buffer drain publish failed, re-enqueueing payloads") - for _, p := range payloads { - if reErr := d.buffer.EnqueueJob(ctx, label, p, d.maxLen); reErr != nil { - d.log.WithError(reErr).Warn("re-enqueue after publish failure failed (data loss possible)") - } + if reErr := d.buffer.EnqueueBatch(ctx, label, payloads, d.maxLen); reErr != nil { + d.log.WithFields(map[string]interface{}{ + "label": label, + "count": len(payloads), + }).WithError(reErr).Warn("re-enqueue after publish failure failed (data loss possible)") } return fmt.Errorf("publish batch %s: %w", topic, pubErr) } diff --git a/pkg/queue/buffering_producer.go b/pkg/queue/buffering_producer.go index 4020d856..4030d093 100644 --- a/pkg/queue/buffering_producer.go +++ b/pkg/queue/buffering_producer.go @@ -13,8 +13,12 @@ import ( // // 본 인터페이스는 BufferingProducer 와 BufferDrainer 가 공유 — Redis 직접 의존 회피. // Noop 구현체 (NoopJobBuffer) 를 wiring 에서 사용하면 buffer 기능이 자동 비활성. +// +// EnqueueBatch (gemini PR #511 피드백): N개 payload 를 단일 Redis pipeline 으로 enqueue — +// PublishBatch / drainer 재적재 경로의 round-trip 부하 N→1 절감. type JobBuffer interface { EnqueueJob(ctx context.Context, label string, payload []byte, maxLen int64) error + EnqueueBatch(ctx context.Context, label string, payloads [][]byte, maxLen int64) error DrainJobs(ctx context.Context, label string, n int) ([][]byte, error) JobBufferLen(ctx context.Context, label string) (int64, error) } @@ -28,6 +32,11 @@ func (NoopJobBuffer) EnqueueJob(_ context.Context, _ string, _ []byte, _ int64) return errors.New("noop job buffer: enqueue not supported") } +// EnqueueBatch 도 동등하게 error — fallback 신호. +func (NoopJobBuffer) EnqueueBatch(_ context.Context, _ string, _ [][]byte, _ int64) error { + return errors.New("noop job buffer: enqueue batch not supported") +} + // DrainJobs 항상 빈 슬라이스 반환 — drainer 가 매 tick 마다 idle 로 인식. func (NoopJobBuffer) DrainJobs(_ context.Context, _ string, _ int) ([][]byte, error) { return nil, nil @@ -124,9 +133,11 @@ func (p *BufferingProducer) Publish(ctx context.Context, msg Message) error { // PublishBatch 는 메시지 단위로 routing 합니다 — 같은 batch 안 high/normal/low 혼합 가능. // -// 구현 — buffer-target 과 direct-target 을 두 슬라이스로 분리 후: -// 1. buffer-target: 순차 EnqueueJob (실패 시 direct 로 재분류) -// 2. direct-target: underlying.PublishBatch 1회 +// 구현 (gemini PR #511 피드백 — 순차 EnqueueJob 회피): +// 1. label 별 (normal/low) payload 슬라이스 + direct-target 슬라이스로 분리 +// 2. 각 label 에 대해 1회 EnqueueBatch 호출 — Redis 1 RTT (multi-arg LPUSH) +// 3. label enqueue 실패 시 해당 label 의 모든 메시지를 direct 로 재분류 — Redis 장애 fallback +// 4. direct-target: underlying.PublishBatch 1회 // // 모든 enqueue 가 성공해도 direct-target 의 PublishBatch 가 실패하면 본 batch 가 부분 성공 — // 호출자는 본 batch 단위의 transactional 보장을 기대하지 않아야 함 (KafkaProducer 와 동일). @@ -135,7 +146,14 @@ func (p *BufferingProducer) PublishBatch(ctx context.Context, msgs []Message) er return nil } + // label 별 buffered payload + 원본 message (enqueue 실패 시 direct 로 재분류) 모음. + type bufferedItem struct { + msg Message + payload []byte + } + byLabel := make(map[string][]bufferedItem) direct := make([]Message, 0, len(msgs)) + for _, msg := range msgs { label := labelForTopic(msg.Topic) if label == "" { @@ -151,13 +169,23 @@ func (p *BufferingProducer) PublishBatch(ctx context.Context, msgs []Message) er direct = append(direct, msg) continue } - if err := p.buffer.EnqueueJob(ctx, label, payload, p.maxLen); err != nil { + byLabel[label] = append(byLabel[label], bufferedItem{msg: msg, payload: payload}) + } + + // label 별 1회 EnqueueBatch — Redis round-trip N→1 절감. + for label, items := range byLabel { + payloads := make([][]byte, len(items)) + for i, it := range items { + payloads[i] = it.payload + } + if err := p.buffer.EnqueueBatch(ctx, label, payloads, p.maxLen); err != nil { p.log.WithFields(map[string]interface{}{ - "topic": msg.Topic, "label": label, - }).WithError(err).Warn("buffer enqueue failed, falling back to direct publish") - direct = append(direct, msg) - continue + "count": len(payloads), + }).WithError(err).Warn("buffer enqueue batch failed, falling back to direct publish") + for _, it := range items { + direct = append(direct, it.msg) + } } } diff --git a/pkg/redis/job_buffer.go b/pkg/redis/job_buffer.go index be761edf..7db95e98 100644 --- a/pkg/redis/job_buffer.go +++ b/pkg/redis/job_buffer.go @@ -32,30 +32,52 @@ func JobBufferKey(label string) string { // 항목이 소실되지만, 정책상 신규 publish 우선. 호출자는 enqueue 직전 IngestionLock 로 dedup 책임. // // label / payload 빈 값은 명시적 error — silent corruption 회피. +// +// 단일 payload 의 thin wrapper — 내부에서 EnqueueBatch 로 위임 (single source of truth). func (c *Client) EnqueueJob(ctx context.Context, label string, payload []byte, maxLen int64) error { - if label == "" { - return errors.New("enqueue job: empty label") - } if len(payload) == 0 { return fmt.Errorf("enqueue job %s: empty payload", label) } + return c.EnqueueBatch(ctx, label, [][]byte{payload}, maxLen) +} + +// EnqueueBatch 는 payloads 슬라이스 전체를 단일 Redis pipeline 으로 LPUSH 합니다 (gemini PR #511 피드백). +// +// 동작: +// 1. 빈 payloads → 즉시 (nil, nil) 반환 (no-op, idempotent) +// 2. 모든 payloads 를 LPUSH 의 variadic args 로 1 RTT 전송 +// 3. MaxLen > 0 이면 동일 pipeline 에 LTRIM 추가 — 2 명령 1 RTT +// +// LPUSH 의 multi-arg 의미: LPUSH key v1 v2 v3 → list head 가 [v3, v2, v1, ...prev] 순서. +// 즉 같은 호출 안의 마지막 인자가 head 최우선. 본 함수는 입력 순서 보존을 목표로 하지 않음 — +// drainer 가 batch 단위로 처리하면 충분. +// +// label 빈 값 / 개별 payload 빈 값 → 명시적 error (silent corruption 회피). +// payloads 길이 0 은 정상 (no-op). +func (c *Client) EnqueueBatch(ctx context.Context, label string, payloads [][]byte, maxLen int64) error { + if label == "" { + return errors.New("enqueue batch: empty label") + } + if len(payloads) == 0 { + return nil + } key := JobBufferKey(label) + args := make([]interface{}, len(payloads)) + for i, p := range payloads { + if len(p) == 0 { + return fmt.Errorf("enqueue batch %s: empty payload at index %d", label, i) + } + args[i] = p + } + pipe := c.rdb.Pipeline() + pipe.LPush(ctx, key, args...) if maxLen > 0 { - // LPUSH + LTRIM 을 pipeline 으로 1 RTT — 두 명령 사이 race 가 있어도 다음 LTRIM 에서 보정. - pipe := c.rdb.Pipeline() - pipe.LPush(ctx, key, payload) - // LTRIM start stop 은 inclusive — [0, maxLen-1] 보존 (head 최신 N개). pipe.LTrim(ctx, key, 0, maxLen-1) - if _, err := pipe.Exec(ctx); err != nil { - return fmt.Errorf("enqueue job %s: %w", label, err) - } - return nil } - - if err := c.rdb.LPush(ctx, key, payload).Err(); err != nil { - return fmt.Errorf("enqueue job %s: %w", label, err) + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("enqueue batch %s: %w", label, err) } return nil } diff --git a/test/internal/scheduler/buffer_drainer_test.go b/test/internal/scheduler/buffer_drainer_test.go index 77e35a6b..f8d428be 100644 --- a/test/internal/scheduler/buffer_drainer_test.go +++ b/test/internal/scheduler/buffer_drainer_test.go @@ -42,6 +42,16 @@ func (b *mockJobBuffer) EnqueueJob(_ context.Context, label string, payload []by return nil } +func (b *mockJobBuffer) EnqueueBatch(_ context.Context, label string, payloads [][]byte, _ int64) error { + if b.enqErr != nil { + return b.enqErr + } + b.mu.Lock() + defer b.mu.Unlock() + b.store[label] = append(b.store[label], payloads...) + return nil +} + func (b *mockJobBuffer) DrainJobs(_ context.Context, label string, n int) ([][]byte, error) { if b.drainErr != nil { return nil, b.drainErr diff --git a/test/pkg/queue/buffering_producer_test.go b/test/pkg/queue/buffering_producer_test.go index a435c1c1..a32fd7bb 100644 --- a/test/pkg/queue/buffering_producer_test.go +++ b/test/pkg/queue/buffering_producer_test.go @@ -69,6 +69,16 @@ func (b *fakeBuffer) EnqueueJob(_ context.Context, label string, payload []byte, return nil } +func (b *fakeBuffer) EnqueueBatch(_ context.Context, label string, payloads [][]byte, _ int64) error { + if b.enqueueErr != nil { + return b.enqueueErr + } + b.mu.Lock() + defer b.mu.Unlock() + b.enqueued[label] = append(b.enqueued[label], payloads...) + return nil +} + func (b *fakeBuffer) DrainJobs(_ context.Context, label string, n int) ([][]byte, error) { b.mu.Lock() defer b.mu.Unlock() @@ -177,6 +187,49 @@ func TestBufferingProducer_Publish_BufferFailFallback(t *testing.T) { assert.Equal(t, 1, under.count(), "buffer 실패 시 underlying 으로 fallback") } +// countingFakeBuffer 는 EnqueueBatch 호출 횟수를 추적해 단일 batch 호출 검증용. +type countingFakeBuffer struct { + *fakeBuffer + batchCallsPerLabel map[string]int +} + +func newCountingFakeBuffer() *countingFakeBuffer { + return &countingFakeBuffer{ + fakeBuffer: newFakeBuffer(), + batchCallsPerLabel: make(map[string]int), + } +} + +func (b *countingFakeBuffer) EnqueueBatch(ctx context.Context, label string, payloads [][]byte, maxLen int64) error { + b.mu.Lock() + b.batchCallsPerLabel[label]++ + b.mu.Unlock() + return b.fakeBuffer.EnqueueBatch(ctx, label, payloads, maxLen) +} + +// TestBufferingProducer_PublishBatch_UsesSingleEnqueueBatchPerLabel 는 같은 label 의 N개 메시지가 +// 단일 EnqueueBatch 호출로 처리되는지 검증 (gemini PR #511 피드백). +func TestBufferingProducer_PublishBatch_UsesSingleEnqueueBatchPerLabel(t *testing.T) { + under := &fakeProducer{} + buf := newCountingFakeBuffer() + p := queue.NewBufferingProducer(under, buf, 1000, newTestLog()) + + msgs := []queue.Message{ + {Topic: queue.TopicCrawlNormal, Value: []byte("n1")}, + {Topic: queue.TopicCrawlNormal, Value: []byte("n2")}, + {Topic: queue.TopicCrawlNormal, Value: []byte("n3")}, + {Topic: queue.TopicCrawlLow, Value: []byte("l1")}, + {Topic: queue.TopicCrawlLow, Value: []byte("l2")}, + } + require.NoError(t, p.PublishBatch(context.Background(), msgs)) + + // 같은 label 의 3개 normal 메시지 → 1회 EnqueueBatch, 2개 low → 1회 EnqueueBatch. + assert.Equal(t, 1, buf.batchCallsPerLabel["normal"], "label 'normal' 1회 batch 호출") + assert.Equal(t, 1, buf.batchCallsPerLabel["low"], "label 'low' 1회 batch 호출") + assert.Equal(t, 3, buf.lenFor("normal")) + assert.Equal(t, 2, buf.lenFor("low")) +} + // TestBufferingProducer_PublishBatch_MixedPriorities 는 batch 안 high/normal/low 혼합 라우팅 검증. func TestBufferingProducer_PublishBatch_MixedPriorities(t *testing.T) { under := &fakeProducer{} diff --git a/test/pkg/redis/job_buffer_test.go b/test/pkg/redis/job_buffer_test.go index 0b86d58d..74fcb3b0 100644 --- a/test/pkg/redis/job_buffer_test.go +++ b/test/pkg/redis/job_buffer_test.go @@ -99,6 +99,73 @@ func TestEnqueueJob_EmptyValidation(t *testing.T) { assert.Error(t, client.EnqueueJob(ctx, "ok", []byte{}, 0)) } +// TestEnqueueBatch_AddsAllPayloadsInOneCall 는 EnqueueBatch 가 multi-arg LPUSH 로 N개를 한 번에 적재 검증. +func TestEnqueueBatch_AddsAllPayloadsInOneCall(t *testing.T) { + client := newTestClient(t) + label := "test-batch" + jobBufferCleanup(t, client, label) + ctx := context.Background() + + payloads := [][]byte{ + []byte("batch-1"), + []byte("batch-2"), + []byte("batch-3"), + } + require.NoError(t, client.EnqueueBatch(ctx, label, payloads, 0)) + + n, err := client.JobBufferLen(ctx, label) + require.NoError(t, err) + assert.Equal(t, int64(3), n) + + drained, err := client.DrainJobs(ctx, label, 10) + require.NoError(t, err) + require.Len(t, drained, 3) +} + +// TestEnqueueBatch_EmptyPayloadsNoop 는 빈 슬라이스가 정상 처리됨을 검증. +func TestEnqueueBatch_EmptyPayloadsNoop(t *testing.T) { + client := newTestClient(t) + label := "test-batch-empty" + jobBufferCleanup(t, client, label) + ctx := context.Background() + + require.NoError(t, client.EnqueueBatch(ctx, label, nil, 0)) + require.NoError(t, client.EnqueueBatch(ctx, label, [][]byte{}, 0)) + + n, err := client.JobBufferLen(ctx, label) + require.NoError(t, err) + assert.Equal(t, int64(0), n) +} + +// TestEnqueueBatch_RespectsLTRIM 는 MaxLen 이 batch 결과에도 적용됨을 검증. +func TestEnqueueBatch_RespectsLTRIM(t *testing.T) { + client := newTestClient(t) + label := "test-batch-trim" + jobBufferCleanup(t, client, label) + ctx := context.Background() + + payloads := make([][]byte, 5) + for i := range payloads { + payloads[i] = []byte("p-" + string(rune('a'+i))) + } + require.NoError(t, client.EnqueueBatch(ctx, label, payloads, 3)) + + n, err := client.JobBufferLen(ctx, label) + require.NoError(t, err) + assert.Equal(t, int64(3), n, "MaxLen=3 로 LTRIM 적용됨") +} + +// TestEnqueueBatch_EmptyPayloadInSliceFails 는 batch 중 빈 payload 가 있으면 error 검증. +func TestEnqueueBatch_EmptyPayloadInSliceFails(t *testing.T) { + client := newTestClient(t) + label := "test-batch-invalid" + jobBufferCleanup(t, client, label) + ctx := context.Background() + + err := client.EnqueueBatch(ctx, label, [][]byte{[]byte("ok"), nil, []byte("ok2")}, 0) + assert.Error(t, err) +} + // TestJobBufferLen 은 enqueue 후 len 이 정확한지 검증합니다. func TestJobBufferLen(t *testing.T) { client := newTestClient(t) From a0b40098762a5b87cab78bb85af4f4e6090fdeeb Mon Sep 17 00:00:00 2001 From: juhyuni Date: Mon, 18 May 2026 23:16:19 +0900 Subject: [PATCH 7/7] =?UTF-8?q?[FIX]:=20=ED=94=BC=EB=93=9C=EB=B0=B1=20?= =?UTF-8?q?=EB=B0=98=EC=98=81,=20buffer=20=ED=99=9C=EC=84=B1=20=EC=8B=9C?= =?UTF-8?q?=20BacklogThrottler=20skip=20+=20drainer=20ctx=20fail-safe=20?= =?UTF-8?q?=EC=99=B8=20(coderabbit=20+=20Copilot)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - (#511 핵심 결함, Copilot) BufferingProducer 활성 시 scheduler BacklogThrottler wiring 자체 skip - 기존 throttler 는 publish 직전 normal/low job drop → BufferingProducer 가 routing 받지도 못함 - bufferDrainer != nil 시 throttle 무시 + INFO 로그 'publisher redis buffer manages elastic queueing' - buffer 가 elastic queueing 책임 인수 — 본 PR 의 핵심 목적 달성 보장 - (coderabbit major) BufferDrainer re-enqueue 경로에 context.WithoutCancel + 5s timeout - shutdown 시 ctx cancel 됐어도 재적재 시도 → 데이터 손실 회피 - (coderabbit major + Copilot) BufferingProducer 생성자 nil guard (underlying / log 필수) - (Copilot) NoopJobBuffer doc 정리 — EnqueueJob/Batch 가 error 반환하는 의도 명시 - (Copilot) TestJobBufferLen t.Cleanup 추가 — 잔존 buffer 가 후속 테스트 영향 회피 - (Copilot) BufferDrainer low priority 테스트 추가 — drainTargets 의 low 매핑 회귀 catch - (Copilot) LoadJobBuffer env 테스트 신규 (default / override / invalid 8 케이스) Co-Authored-By: Claude Opus 4.7 (1M context) --- cmd/issuetracker/main.go | 12 ++- internal/scheduler/buffer_drainer.go | 8 +- pkg/queue/buffering_producer.go | 28 ++++-- .../internal/scheduler/buffer_drainer_test.go | 33 +++++++ test/pkg/config/config_test.go | 87 +++++++++++++++++++ test/pkg/redis/job_buffer_test.go | 3 + 6 files changed, 160 insertions(+), 11 deletions(-) diff --git a/cmd/issuetracker/main.go b/cmd/issuetracker/main.go index 0fb67db5..48e2f715 100644 --- a/cmd/issuetracker/main.go +++ b/cmd/issuetracker/main.go @@ -862,7 +862,17 @@ func main() { // Backlog throttle: SCHEDULER_MAX_BACKLOG > 0 일 때만 활성. // crawl 토픽의 consumer-group lag 가 임계값 초과 시 publish 차단. - if schedulerCfg.MaxBacklog > 0 { + // + // 이슈 #510 — Redis 버퍼 활성 시 throttle wire skip: + // - BacklogThrottler 는 scheduler.runEntry 에서 publish 직전 normal/low job 을 silent drop + // - 본 PR 의 BufferingProducer 는 publish 단계에서 normal/low 를 Redis 로 routing + // - 둘 다 활성화하면 throttle drop 이 BufferingProducer 진입 전 발생 → buffer 가 손실 경로 우회 못함 (Copilot PR #511 피드백) + // 따라서 buffer 활성 시 throttler 자체를 wiring 하지 않음 — buffer 가 elastic queueing 책임 인수. + switch { + case bufferDrainer != nil: + log.WithField("max_backlog_ignored", schedulerCfg.MaxBacklog). + Info("scheduler backlog throttle skipped — publisher redis buffer manages elastic queueing") + case schedulerCfg.MaxBacklog > 0: backlogChecker := queue.NewBacklogChecker(crawlerKafkaCfg.Brokers, schedulerCfg.BacklogCheckTimeout) throttler := scheduler.NewBacklogThrottler( backlogChecker, diff --git a/internal/scheduler/buffer_drainer.go b/internal/scheduler/buffer_drainer.go index 82b8afdc..27c51a83 100644 --- a/internal/scheduler/buffer_drainer.go +++ b/internal/scheduler/buffer_drainer.go @@ -237,12 +237,18 @@ func (d *BufferDrainer) drainOnce(ctx context.Context, label, topic string) erro if pubErr := d.producer.PublishBatch(ctx, msgs); pubErr != nil { // publish 실패 → 재적재 (순서 보존 X). EnqueueBatch 로 1 RTT — N개 EnqueueJob 회피 // (gemini PR #511 피드백). + // + // shutdown 중 ctx 가 cancel 됐어도 재적재는 반드시 시도해야 데이터 손실 회피 — + // context.WithoutCancel 로 logger / trace metadata 는 보존하되 cancel 신호 분리 (coderabbit + // PR #511 피드백). 5s timeout 으로 bound — Redis stall 시 goroutine leak 회피. d.log.WithFields(map[string]interface{}{ "label": label, "topic": topic, "count": len(msgs), }).WithError(pubErr).Warn("buffer drain publish failed, re-enqueueing payloads") - if reErr := d.buffer.EnqueueBatch(ctx, label, payloads, d.maxLen); reErr != nil { + reCtx, reCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer reCancel() + if reErr := d.buffer.EnqueueBatch(reCtx, label, payloads, d.maxLen); reErr != nil { d.log.WithFields(map[string]interface{}{ "label": label, "count": len(payloads), diff --git a/pkg/queue/buffering_producer.go b/pkg/queue/buffering_producer.go index 4030d093..cb9a5ba7 100644 --- a/pkg/queue/buffering_producer.go +++ b/pkg/queue/buffering_producer.go @@ -23,8 +23,14 @@ type JobBuffer interface { JobBufferLen(ctx context.Context, label string) (int64, error) } -// NoopJobBuffer 는 모든 호출이 zero-value 를 반환하는 fail-safe 구현체입니다. -// Redis 비활성 / buffer 기능 opt-out 시 wiring 에서 사용. +// NoopJobBuffer 는 fail-safe 구현체입니다 — Redis 비활성 / buffer 기능 opt-out 시 wiring 사용. +// +// 동작: +// - EnqueueJob / EnqueueBatch : 항상 error 반환 → BufferingProducer 가 underlying 직접 publish fallback +// - DrainJobs / JobBufferLen : 빈 슬라이스 / 0 반환 (idle) +// +// enqueue 가 error 를 반환하는 이유: BufferingProducer 의 fallback 경로 진입 신호 — +// silent succeed 시 payload 가 사라진 채로 publish 손실 발생. 명시적 error 만이 fallback 트리거. type NoopJobBuffer struct{} // EnqueueJob 항상 error 반환 — BufferingProducer 가 fallback 경로를 타도록 신호. @@ -71,14 +77,18 @@ type BufferingProducer struct { // NewBufferingProducer 는 underlying Producer 를 buffer 데코레이터로 감쌉니다. // // 인자: -// - underlying : 실제 Kafka publish 책임 (보통 *KafkaProducer) -// - buffer : Redis 또는 Noop. nil 이면 NoopJobBuffer 로 자동 대체 — 기능 비활성 -// - maxLen : EnqueueJob 의 LIST 최대 길이 (>0 이면 LTRIM 적용) -// - log : enqueue 실패 / fallback 등 WARN 로그 -// -// underlying / log 는 필수 (nil 이면 NewBufferingProducer 자체가 fallback 으로 underlying 만 반환할 수 -// 있으나, 그러면 데코 의미 없음 → 명시적 error 또는 호출자가 wiring 책임). +// - underlying : 실제 Kafka publish 책임 (보통 *KafkaProducer). nil 이면 panic — exported +// constructor 이므로 wiring 오류는 silent crash 대신 즉시 명시 (coderabbit PR #511 피드백). +// - buffer : Redis 또는 Noop. nil 이면 NoopJobBuffer 로 자동 대체 — 기능 비활성 의미. +// - maxLen : EnqueueJob/Batch 의 LIST 최대 길이 (>0 이면 LTRIM 적용) +// - log : enqueue 실패 / fallback 등 WARN 로그. nil 이면 panic (필수). func NewBufferingProducer(underlying Producer, buffer JobBuffer, maxLen int64, log *logger.Logger) *BufferingProducer { + if underlying == nil { + panic("queue: NewBufferingProducer requires non-nil underlying Producer") + } + if log == nil { + panic("queue: NewBufferingProducer requires non-nil logger") + } if buffer == nil { buffer = NoopJobBuffer{} } diff --git a/test/internal/scheduler/buffer_drainer_test.go b/test/internal/scheduler/buffer_drainer_test.go index f8d428be..ce92733f 100644 --- a/test/internal/scheduler/buffer_drainer_test.go +++ b/test/internal/scheduler/buffer_drainer_test.go @@ -277,6 +277,39 @@ func TestBufferDrainer_IdleWhenBufferEmpty(t *testing.T) { assert.Equal(t, 0, prod.count()) } +// TestBufferDrainer_DrainsLowPriority 는 low label 도 정상 drain 되는지 검증 (Copilot PR #511 피드백). +// normal 만 검증하면 drainTargets 의 low 매핑 누락/오류가 회귀로 잡히지 않음. +func TestBufferDrainer_DrainsLowPriority(t *testing.T) { + buf := newMockJobBuffer() + for i := 0; i < 3; i++ { + payload := encodeMsg(t, queue.Message{Topic: queue.TopicCrawlLow, Value: []byte("v")}) + buf.seed("low", payload) + } + + prod := &mockDrainerProducer{} + check := &fixedBacklogChecker{} + check.lag.Store(50) + + d, err := scheduler.NewBufferDrainer(buf, prod, check, scheduler.BufferDrainerConfig{ + Interval: time.Hour, + TargetBacklog: 1000, + DrainBatch: 100, + GroupID: queue.GroupCrawlerWorkers, + }, newDrainerTestLog()) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + d.Start(ctx) + time.Sleep(80 * time.Millisecond) + cancel() + d.Stop() + + assert.Equal(t, 3, prod.count(), "low priority buffer 3건 모두 drain") + for _, m := range prod.published { + assert.Equal(t, queue.TopicCrawlLow, m.Topic, "low buffer payload 는 low topic 으로 publish") + } +} + // TestBufferDrainer_NilDepsReturnError 는 필수 의존성 nil 시 생성자가 error 반환을 검증. func TestBufferDrainer_NilDepsReturnError(t *testing.T) { buf := newMockJobBuffer() diff --git a/test/pkg/config/config_test.go b/test/pkg/config/config_test.go index 430fd917..a9d88222 100644 --- a/test/pkg/config/config_test.go +++ b/test/pkg/config/config_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -510,6 +511,92 @@ func TestLoadScheduler_InvalidBacklogCheckTimeout(t *testing.T) { } } +// ───────────────────────────────────────────────────────────────────────────── +// LoadJobBuffer (이슈 #510 / PR #511 Copilot 피드백) +// ───────────────────────────────────────────────────────────────────────────── + +func unsetJobBufferEnvVars(t *testing.T) { + t.Helper() + vars := []string{ + "PUBLISHER_REDIS_BUFFER_ENABLED", + "PUBLISHER_REDIS_BUFFER_DRAIN_INTERVAL", + "PUBLISHER_REDIS_BUFFER_TARGET_BACKLOG", + "PUBLISHER_REDIS_BUFFER_DRAIN_BATCH", + "PUBLISHER_REDIS_BUFFER_MAX_LEN", + "PUBLISHER_REDIS_BUFFER_CHECK_TIMEOUT", + } + for _, v := range vars { + t.Setenv(v, "") + } +} + +func TestLoadJobBuffer_DefaultValues(t *testing.T) { + unsetJobBufferEnvVars(t) + + cfg, err := processorcfg.LoadJobBuffer("/tmp/nonexistent-env-file.env") + require.NoError(t, err) + + def := processorcfg.DefaultJobBufferConfig() + if cfg.Enabled != def.Enabled { + t.Errorf("Enabled: got %v, want %v (default disabled)", cfg.Enabled, def.Enabled) + } + if cfg.DrainInterval != def.DrainInterval { + t.Errorf("DrainInterval: got %v, want %v", cfg.DrainInterval, def.DrainInterval) + } + if cfg.TargetBacklog != def.TargetBacklog { + t.Errorf("TargetBacklog: got %d, want %d", cfg.TargetBacklog, def.TargetBacklog) + } + if cfg.DrainBatch != def.DrainBatch { + t.Errorf("DrainBatch: got %d, want %d", cfg.DrainBatch, def.DrainBatch) + } + if cfg.MaxLen != def.MaxLen { + t.Errorf("MaxLen: got %d, want %d", cfg.MaxLen, def.MaxLen) + } +} + +func TestLoadJobBuffer_EnvOverride(t *testing.T) { + unsetJobBufferEnvVars(t) + t.Setenv("PUBLISHER_REDIS_BUFFER_ENABLED", "true") + t.Setenv("PUBLISHER_REDIS_BUFFER_DRAIN_INTERVAL", "10s") + t.Setenv("PUBLISHER_REDIS_BUFFER_TARGET_BACKLOG", "1500") + t.Setenv("PUBLISHER_REDIS_BUFFER_DRAIN_BATCH", "50") + t.Setenv("PUBLISHER_REDIS_BUFFER_MAX_LEN", "0") // 0 = unlimited (NonNegativeInt64) + + cfg, err := processorcfg.LoadJobBuffer("/tmp/nonexistent-env-file.env") + require.NoError(t, err) + + assert.True(t, cfg.Enabled) + assert.Equal(t, 10*time.Second, cfg.DrainInterval) + assert.Equal(t, int64(1500), cfg.TargetBacklog) + assert.Equal(t, 50, cfg.DrainBatch) + assert.Equal(t, int64(0), cfg.MaxLen) +} + +func TestLoadJobBuffer_InvalidInputs(t *testing.T) { + tests := []struct { + name string + key string + value string + }{ + {"invalid duration", "PUBLISHER_REDIS_BUFFER_DRAIN_INTERVAL", "not-a-duration"}, + {"non-positive duration", "PUBLISHER_REDIS_BUFFER_DRAIN_INTERVAL", "0s"}, + {"invalid int (target)", "PUBLISHER_REDIS_BUFFER_TARGET_BACKLOG", "not-a-number"}, + {"non-positive target", "PUBLISHER_REDIS_BUFFER_TARGET_BACKLOG", "0"}, + {"invalid int (batch)", "PUBLISHER_REDIS_BUFFER_DRAIN_BATCH", "abc"}, + {"non-positive batch", "PUBLISHER_REDIS_BUFFER_DRAIN_BATCH", "-5"}, + {"negative maxlen", "PUBLISHER_REDIS_BUFFER_MAX_LEN", "-1"}, + {"invalid check timeout", "PUBLISHER_REDIS_BUFFER_CHECK_TIMEOUT", "xyz"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + unsetJobBufferEnvVars(t) + t.Setenv(tt.key, tt.value) + _, err := processorcfg.LoadJobBuffer("/tmp/nonexistent-env-file.env") + assert.Error(t, err, "%s=%q 는 에러를 반환해야 함", tt.key, tt.value) + }) + } +} + // ───────────────────────────────────────────────────────────────────────────── // LoadPathInfer (이슈 #173 단계 2) // ───────────────────────────────────────────────────────────────────────────── diff --git a/test/pkg/redis/job_buffer_test.go b/test/pkg/redis/job_buffer_test.go index 74fcb3b0..90b93eb5 100644 --- a/test/pkg/redis/job_buffer_test.go +++ b/test/pkg/redis/job_buffer_test.go @@ -172,6 +172,9 @@ func TestJobBufferLen(t *testing.T) { label := "test-len" jobBufferCleanup(t, client, label) ctx := context.Background() + // 잔존 buffer 가 후속 테스트 실행 / 디버깅에 영향을 주지 않도록 종료 시 정리 보장 + // (Copilot PR #511 피드백). + t.Cleanup(func() { jobBufferCleanup(t, client, label) }) n, err := client.JobBufferLen(ctx, label) require.NoError(t, err)