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/cmd/issuetracker/main.go b/cmd/issuetracker/main.go
index 23f03515..48e2f715 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()
@@ -815,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,
@@ -845,6 +902,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 +1154,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
diff --git a/internal/scheduler/buffer_drainer.go b/internal/scheduler/buffer_drainer.go
new file mode 100644
index 00000000..27c51a83
--- /dev/null
+++ b/internal/scheduler/buffer_drainer.go
@@ -0,0 +1,269 @@
+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). 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")
+		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),
+			}).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/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
+}
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..cb9a5ba7
--- /dev/null
+++ b/pkg/queue/buffering_producer.go
@@ -0,0 +1,223 @@
+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 기능이 자동 비활성.
+//
+// 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)
+}
+
+// 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 경로를 타도록 신호.
+func (NoopJobBuffer) EnqueueJob(_ context.Context, _ string, _ []byte, _ int64) error {
+	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
+}
+
+// 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). 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{}
+	}
+	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 혼합 가능.
+//
+// 구현 (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 와 동일).
+func (p *BufferingProducer) PublishBatch(ctx context.Context, msgs []Message) error {
+	if len(msgs) == 0 {
+		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 == "" {
+			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
+		}
+		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{}{
+				"label": label,
+				"count": len(payloads),
+			}).WithError(err).Warn("buffer enqueue batch failed, falling back to direct publish")
+			for _, it := range items {
+				direct = append(direct, it.msg)
+			}
+		}
+	}
+
+	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/pkg/redis/job_buffer.go b/pkg/redis/job_buffer.go
new file mode 100644
index 00000000..7db95e98
--- /dev/null
+++ b/pkg/redis/job_buffer.go
@@ -0,0 +1,135 @@
+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:<priority> 형식.
+//
+// 자료구조 — 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 회피.
+//
+// 단일 payload 의 thin wrapper — 내부에서 EnqueueBatch 로 위임 (single source of truth).
+func (c *Client) EnqueueJob(ctx context.Context, label string, payload []byte, maxLen int64) error {
+	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 {
+		pipe.LTrim(ctx, key, 0, maxLen-1)
+	}
+	if _, err := pipe.Exec(ctx); err != nil {
+		return fmt.Errorf("enqueue batch %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/internal/scheduler/buffer_drainer_test.go b/test/internal/scheduler/buffer_drainer_test.go
new file mode 100644
index 00000000..ce92733f
--- /dev/null
+++ b/test/internal/scheduler/buffer_drainer_test.go
@@ -0,0 +1,329 @@
+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) 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
+	}
+	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_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()
+	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")
+}
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/queue/buffering_producer_test.go b/test/pkg/queue/buffering_producer_test.go
new file mode 100644
index 00000000..a32fd7bb
--- /dev/null
+++ b/test/pkg/queue/buffering_producer_test.go
@@ -0,0 +1,302 @@
+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) 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()
+	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")
+}
+
+// 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{}
+	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 시점에 채워짐")
+}
diff --git a/test/pkg/redis/job_buffer_test.go b/test/pkg/redis/job_buffer_test.go
new file mode 100644
index 00000000..90b93eb5
--- /dev/null
+++ b/test/pkg/redis/job_buffer_test.go
@@ -0,0 +1,190 @@
+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))
+}
+
+// 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)
+	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)
+	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)
+}