Feature: Add recover pending for redis stream_queue - #14
Conversation
📝 WalkthroughWalkthroughWorker configuration now supports claim timing and batch settings. Redis streams can claim idle pending messages, and workers recover them periodically. Ingest and reindex skip already succeeded tasks, while several exported and helper methods receive documentation comments. ChangesWorker recovery and processing
Estimated code review effort: 3 (Moderate) | ~30 minutes Sequence Diagram(s)sequenceDiagram
participant Worker
participant RedisStreamQueue
participant handleBatch
Worker->>RedisStreamQueue: ClaimPending with cursor, idle threshold, and batch size
RedisStreamQueue-->>Worker: Claimed messages and next cursor
Worker->>handleBatch: Process claimed messages
handleBatch-->>Worker: Processing result
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
cmd/worker/main.go (1)
87-111: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
cfg.WorkerJobTimeoutis logged but never actually applied —JobTimeoutstays hardcoded.
WorkerOptions.JobTimeoutis set to a literal30 * time.Minute, ignoring the newly-addedcfg.WorkerJobTimeout(sourced fromCERULEAN_WORKER_JOB_TIMEOUTviaenvDurationinconfig.go). The startup log at Line 109 then printscfg.WorkerJobTimeout, giving the false impression that this value governs job execution timeouts — an operator who sets the env var will see it reflected in the log but observe no actual change in behavior.🐛 Proposed fix
worker, err := executor.NewWorker(q, registry, executor.WorkerOptions{ BatchSize: cfg.WorkerBatchSize, BlockMillis: 5000, - JobTimeout: 30 * time.Minute, + JobTimeout: cfg.WorkerJobTimeout, Concurrency: cfg.WorkerConcurrency,🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cmd/worker/main.go` around lines 87 - 111, Update the WorkerOptions initialization in the executor.NewWorker call to assign JobTimeout from cfg.WorkerJobTimeout instead of the hardcoded 30-minute duration, keeping the existing startup log consistent with the applied configuration.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/executor/worker.go`:
- Around line 122-141: Initialize lastClaimAt so the first loop iteration
satisfies the recovery condition, allowing recoverPending to run immediately
when the worker starts. Update the initialization adjacent to the worker loop
while preserving the existing periodic claimInterval scheduling and error
handling.
In `@internal/ingest/service.go`:
- Around line 379-381: Correct the call to ProcessPaperIngest in the paper
ingest pipeline to pass arguments in its declared order: context, paperID, then
taskID. Keep the reindex handler’s existing argument order unchanged so normal
ingest and recovery resolve the intended paper and task before relying on the
succeeded-status guard.
In `@internal/queue/redis_stream.go`:
- Around line 150-152: Update the minIdle validation in the surrounding public
queue method to use a non-positive check, so minIdle == 0 receives the existing
35-minute default just like negative values; leave the neighboring max guard and
other behavior unchanged.
- Around line 132-138: Update decodeRedisMessages and its
DequeueBatch/ClaimPending callers to decode messages best-effort: retain
successfully decoded entries while handling malformed payloads individually, and
ensure failures cannot prevent cursor advancement or block later messages.
Preserve terminal-error handling by logging/reporting each invalid entry, or
explicitly acknowledging entries that cannot be processed.
- Around line 193-196: Implement failure handling in RedisStreamQueue.Nack:
increment and persist the job’s attempt count, enforce a retry limit, and route
messages exceeding that limit to the existing dead-letter path instead of
leaving them eligible for recoverPending indefinitely. Use the Job.Attempt field
and existing queue acknowledgment/dead-letter symbols, preserving normal retry
behavior below the cap.
---
Outside diff comments:
In `@cmd/worker/main.go`:
- Around line 87-111: Update the WorkerOptions initialization in the
executor.NewWorker call to assign JobTimeout from cfg.WorkerJobTimeout instead
of the hardcoded 30-minute duration, keeping the existing startup log consistent
with the applied configuration.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 3d102b46-1d14-4dc2-b7fb-0d0ce0c0d1d6
📒 Files selected for processing (11)
cmd/worker/main.gointernal/config/config.gointernal/executor/registry.gointernal/executor/worker.gointernal/ingest/service.gointernal/parser/chunker.gointernal/parser/pdf_text.gointernal/pipeline/paper_ingest.gointernal/pipeline/paper_reindex.gointernal/queue/queue.gointernal/queue/redis_stream.go
| lastClaimAt := time.Now() | ||
|
|
||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| log.Println("executor worker stopping") | ||
| return ctx.Err() | ||
| default: | ||
| } | ||
|
|
||
| if lastClaimAt.IsZero() || time.Since(lastClaimAt) >= w.claimInterval { | ||
| if err := w.recoverPending(ctx); err != nil { | ||
| if ctx.Err() != nil { | ||
| return ctx.Err() | ||
| } | ||
|
|
||
| log.Printf("recover pending jobs failed: %v", err) | ||
| } | ||
| lastClaimAt = time.Now() | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
First pending-recovery pass is delayed by a full claimInterval after startup.
lastClaimAt := time.Now() is set immediately before the loop, so time.Since(lastClaimAt) >= w.claimInterval is false on the first iteration and recoverPending is skipped until an interval has elapsed. Since a core goal of this feature is recovering jobs left pending by a previously-crashed worker, delaying that first recovery attempt (default 1 minute, configurable) seems counter to intent.
♻️ Proposed fix
- lastClaimAt := time.Now()
+ var lastClaimAt time.Time📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| lastClaimAt := time.Now() | |
| for { | |
| select { | |
| case <-ctx.Done(): | |
| log.Println("executor worker stopping") | |
| return ctx.Err() | |
| default: | |
| } | |
| if lastClaimAt.IsZero() || time.Since(lastClaimAt) >= w.claimInterval { | |
| if err := w.recoverPending(ctx); err != nil { | |
| if ctx.Err() != nil { | |
| return ctx.Err() | |
| } | |
| log.Printf("recover pending jobs failed: %v", err) | |
| } | |
| lastClaimAt = time.Now() | |
| } | |
| var lastClaimAt time.Time | |
| for { | |
| select { | |
| case <-ctx.Done(): | |
| log.Println("executor worker stopping") | |
| return ctx.Err() | |
| default: | |
| } | |
| if lastClaimAt.IsZero() || time.Since(lastClaimAt) >= w.claimInterval { | |
| if err := w.recoverPending(ctx); err != nil { | |
| if ctx.Err() != nil { | |
| return ctx.Err() | |
| } | |
| log.Printf("recover pending jobs failed: %v", err) | |
| } | |
| lastClaimAt = time.Now() | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/executor/worker.go` around lines 122 - 141, Initialize lastClaimAt
so the first loop iteration satisfies the recovery condition, allowing
recoverPending to run immediately when the worker starts. Update the
initialization adjacent to the worker loop while preserving the existing
periodic claimInterval scheduling and error handling.
| if job.Status == task.Succeeded { | ||
| return nil | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Fix the ingest caller’s argument order before relying on this guard.
ProcessPaperIngest expects (ctx, paperID, taskID), but Line 37 of internal/pipeline/paper_ingest.go passes (ctx, taskID, paperID); the reindex handler uses the correct order. Normal ingest and recovery therefore look up the wrong task and paper, preventing this succeeded-task guard from applying to the intended job.
Proposed fix
- return h.ingest.ProcessPaperIngest(ctx, taskID, paperID)
+ return h.ingest.ProcessPaperIngest(ctx, paperID, taskID)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/ingest/service.go` around lines 379 - 381, Correct the call to
ProcessPaperIngest in the paper ingest pipeline to pass arguments in its
declared order: context, paperID, then taskID. Keep the reindex handler’s
existing argument order unchanged so normal ingest and recovery resolve the
intended paper and task before relying on the succeeded-status guard.
| redisMessages := make([]redis.XMessage, 0) | ||
|
|
||
| for _, stream := range streams { | ||
| for _, redisMsg := range stream.Messages { | ||
| raw, ok := redisMsg.Values["payload"].(string) | ||
| if !ok { | ||
| continue | ||
| } | ||
|
|
||
| var job Job | ||
| if err := json.Unmarshal([]byte(raw), &job); err != nil { | ||
| continue | ||
| } | ||
|
|
||
| messages = append(messages, Message{ | ||
| RedisID: redisMsg.ID, | ||
| Job: job, | ||
| }) | ||
| redisMessages = append(redisMessages, stream.Messages...) | ||
| } | ||
| return decodeRedisMessages(redisMessages) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Single malformed payload discards the whole decoded batch.
decodeRedisMessages returns nil, err on the first bad payload, discarding any already-decoded messages in that batch for the claimed entries as an array along with a stream ID intended for cursor-like use as the start argument for its subsequent call. Per the change details, this replaces prior behavior that skip invalid payloads/unmarshal errors.
Concretely:
- In
DequeueBatch, one bad message causes the whole read batch to be dropped from processing (messages stay unacked in the PEL, delayed but not lost). - In
ClaimPending, since the cursor (nextStart) is only advanced by the caller whenerr == nil(seeworker.gorecoverPending), a poison message at the head of a claimed batch causesrecoverPendingto fail every cycle without ever advancing past it — and sinceXAUTOCLAIMresets idle time on claim, the message won't be eligible for reclaim again untilminIdleelapses, but it will keep resurfacing at the same cursor position, permanently blocking the batch behind it from being recovered.
Consider decoding messages best-effort (collecting successfully-decoded messages and reporting/logging failures individually, or explicitly acking terminally-bad entries) instead of failing the entire batch.
Also applies to: 158-177, 206-238
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/queue/redis_stream.go` around lines 132 - 138, Update
decodeRedisMessages and its DequeueBatch/ClaimPending callers to decode messages
best-effort: retain successfully decoded entries while handling malformed
payloads individually, and ensure failures cannot prevent cursor advancement or
block later messages. Preserve terminal-error handling by logging/reporting each
invalid entry, or explicitly acknowledging entries that cannot be processed.
| if minIdle < 0 { | ||
| minIdle = 35 * time.Minute | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
minIdle guard should use <= 0 for consistency with the max guard.
if minIdle < 0 {
minIdle = 35 * time.Minute
}A caller passing minIdle == 0 slips through un-defaulted (unlike max, guarded with <= 0 two lines below), and 0 means "claim regardless of idle time," which would let this consumer steal messages that another consumer is still actively processing. Even though the current in-repo caller (Worker.recoverPending) always supplies a validated non-zero value, this is a public interface method and the asymmetric guard is a latent footgun.
🛡️ Proposed fix
- if minIdle < 0 {
+ if minIdle <= 0 {
minIdle = 35 * time.Minute
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/queue/redis_stream.go` around lines 150 - 152, Update the minIdle
validation in the surrounding public queue method to use a non-positive check,
so minIdle == 0 receives the existing 35-minute default just like negative
values; leave the neighboring max guard and other behavior unchanged.
| // Nack handles a failed Redis message. | ||
| func (q *RedisStreamQueue) Nack(ctx context.Context, msg Message, reason error) error { | ||
| return nil | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether Job.Attempt is read/incremented anywhere in the codebase
rg -n '\.Attempt\b' --type=goRepository: CeruleanFlow/Cerulean
Length of output: 159
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant queue implementation and job type.
git ls-files | rg '^(internal/queue/redis_stream\.go|internal/queue/.*\.go)$' || true
echo '--- redis_stream.go around Nack / recoverPending / ClaimPending ---'
nl -ba internal/queue/redis_stream.go | sed -n '1,320p'
echo '--- Job definition and attempt usage across queue package ---'
rg -n 'type Job|Attempt|ClaimPending|recoverPending|Nack\(' internal/queue --type=goRepository: CeruleanFlow/Cerulean
Length of output: 344
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find any tests or docs that describe retry/attempt semantics for this queue.
rg -n 'retry|dead[- ]letter|deadletter|Attempt|recoverPending|claimInterval|claimMinIdle' . --type=go --type=md --type=yaml --type=tomlRepository: CeruleanFlow/Cerulean
Length of output: 1823
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the queue package symbols before reading smaller slices if needed.
ast-grep outline internal/queue/redis_stream.go --view expanded || trueRepository: CeruleanFlow/Cerulean
Length of output: 1405
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- internal/queue/job.go ---'
sed -n '1,120p' internal/queue/job.go
echo '--- internal/executor/worker.go around processing and recoverPending ---'
sed -n '220,320p' internal/executor/worker.go
echo '--- internal/queue/redis_stream.go around ClaimPending/Ack/Nack ---'
sed -n '130,210p' internal/queue/redis_stream.goRepository: CeruleanFlow/Cerulean
Length of output: 4976
Add a retry cap or dead-letter path for repeated failures. Nack is still a no-op, and Job.Attempt is never incremented or checked anywhere, so a job that keeps failing will be reclaimed by recoverPending forever and keep burning worker capacity.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/queue/redis_stream.go` around lines 193 - 196, Implement failure
handling in RedisStreamQueue.Nack: increment and persist the job’s attempt
count, enforce a retry limit, and route messages exceeding that limit to the
existing dead-letter path instead of leaving them eligible for recoverPending
indefinitely. Use the Job.Attempt field and existing queue
acknowledgment/dead-letter symbols, preserving normal retry behavior below the
cap.
1 participant
Summary by CodeRabbit
Related to #5
New Features
Bug Fixes
Documentation