Feature: add redis stream queue - #13
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (4)
🚧 Files skipped from review as they are similar to previous changes (2)
📝 WalkthroughWalkthroughThis PR replaces in-process ingest execution with Redis Stream jobs, adds executor and worker runtime behavior, wires queue configuration into server and worker startup, registers pipeline handlers, changes reindex requests to return accepted tasks, and updates Elasticsearch index handling and CI worker builds. ChangesQueued ingest worker pipeline
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant API
participant IngestService
participant RedisStreamQueue
participant Worker
participant PipelineHandler
API->>IngestService: StartPaperIngest or StartPaperReindex
IngestService->>RedisStreamQueue: Enqueue(queue.Job)
API-->>API: Return accepted task
Worker->>RedisStreamQueue: DequeueBatch(...)
RedisStreamQueue-->>Worker: queue.Message
Worker->>PipelineHandler: Handle(ctx, job)
PipelineHandler->>IngestService: ProcessPaperIngest or ProcessPaperReindex
IngestService-->>PipelineHandler: processing result
Worker->>RedisStreamQueue: Ack or Nack(message)
🚥 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
🧹 Nitpick comments (2)
internal/queue/redis_stream.go (2)
87-102: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winStream is never trimmed — unbounded growth risk.
EnqueuecallsXAddwith noMaxLen/Approx(or TTL/MINID) cap, socerulean_taskswill grow without bound as jobs accumulate, consuming ever-increasing Redis memory over time.♻️ Proposed fix
return q.client.XAdd(ctx, &redis.XAddArgs{ Stream: q.stream, + MaxLen: 100_000, + Approx: true, Values: map[string]any{ "payload": string(payload), "type": job.Type, "task_id": job.TaskID, "paper_id": job.PaperID, }, }).Err()🤖 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 87 - 102, The Redis stream in RedisStreamQueue.Enqueue is being appended to without any retention cap, which can cause unbounded growth. Update the q.client.XAdd call in Enqueue to enforce stream trimming using Redis max-length or MINID retention (prefer approximate trimming if acceptable), and make the limit configurable if needed so the stream does not grow forever.
31-71: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winClient is leaked if
ensureGroupfails; noClose()exposed for graceful shutdown.If
ensureGroupfails (Line 66-68), the already-connectedclientis dropped without callingclient.Close(), leaking the connection pool. More generally,RedisStreamQueuenever exposes a way to close the underlying client at all, making graceful shutdown of the worker impossible.♻️ Proposed fix
if err := q.ensureGroup(ctx); err != nil { + client.Close() return nil, err } return q, nil } + +func (q *RedisStreamQueue) Close() error { + return q.client.Close() +}🤖 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 31 - 71, The Redis client created in NewRedisStreamQueue is leaked when q.ensureGroup fails because the connected client is returned from scope without being closed. Update NewRedisStreamQueue to close client on any post-connect failure path, especially after ensureGroup, and add a Close method on RedisStreamQueue that delegates to the underlying redis.Client so callers can shut it down cleanly during worker shutdown.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@cmd/worker/main.go`:
- Line 1: The worker entrypoint is missing from the cmd/worker package, so the
binary cannot build. Add a func main() to the main package in cmd/worker/main.go
and ensure it performs the worker startup path expected by the existing worker
code, using the main package entrypoint as the unique location to fix.
In `@internal/queue/redis_stream.go`:
- Around line 159-161: The RedisStreamQueue Nack implementation is a permanent
no-op, so failed messages are never requeued, retried, or dead-lettered. Update
Nack on RedisStreamQueue to either implement the intended retry/recovery path
(for example via XClaim/XAutoClaim handling in the worker flow) or, if this is
intentionally deferred, add an explicit TODO/documentation in Nack and the Queue
contract so callers know no retry semantics are provided yet.
- Around line 151-157: The Ack method in RedisStreamQueue is passing msg.RedisID
as both the consumer group and the message ID, so the pending entry is never
acknowledged for q.group. Update Ack to call XAck with q.stream, q.group, and
the message RedisID as the ID argument only, keeping the empty RedisID guard
intact.
- Around line 128-149: DequeueBatch in redis_stream.go is silently skipping
malformed Redis stream entries with missing payloads or bad JSON, leaving them
stuck in the consumer PEL. Update the loop that builds messages from
stream.Messages to explicitly handle these failures by logging the bad entry and
either acking it immediately or routing it through the queue’s dead-letter path
instead of continuing past it. Use the existing DequeueBatch flow and the
redisMsg.ID/job parsing logic to ensure malformed records cannot be stranded.
- Around line 104-119: The RedisStreamQueue.DequeueBatch timeout handling only
normalizes negative blockMillis, so a value of 0 still gets passed through to
XReadGroup as BLOCK 0 and blocks forever. Update the guard in DequeueBatch to
treat 0 the same as an unset timeout and fall back to the default 5000ms, while
keeping the existing max normalization and XReadGroup call unchanged.
---
Nitpick comments:
In `@internal/queue/redis_stream.go`:
- Around line 87-102: The Redis stream in RedisStreamQueue.Enqueue is being
appended to without any retention cap, which can cause unbounded growth. Update
the q.client.XAdd call in Enqueue to enforce stream trimming using Redis
max-length or MINID retention (prefer approximate trimming if acceptable), and
make the limit configurable if needed so the stream does not grow forever.
- Around line 31-71: The Redis client created in NewRedisStreamQueue is leaked
when q.ensureGroup fails because the connected client is returned from scope
without being closed. Update NewRedisStreamQueue to close client on any
post-connect failure path, especially after ensureGroup, and add a Close method
on RedisStreamQueue that delegates to the underlying redis.Client so callers can
shut it down cleanly during worker shutdown.
🪄 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: dc3dd6e2-bad2-44b1-8e51-3a3dd00e98b6
⛔ Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
📒 Files selected for processing (11)
cmd/worker/main.gogo.modinternal/config/config.gointernal/executor/executor.gointernal/executor/registry.gointernal/executor/worker.gointernal/pipeline/paper_ingest.gointernal/pipeline/paper_reindex.gointernal/queue/job.gointernal/queue/queue.gointernal/queue/redis_stream.go
| messages := make([]Message, 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, | ||
| }) | ||
| } | ||
| } | ||
| return messages, nil | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the target file and inspect the relevant functions.
ast-grep outline internal/queue/redis_stream.go --view expanded || true
wc -l internal/queue/redis_stream.go
sed -n '1,260p' internal/queue/redis_stream.go
# Search for acknowledgement / recovery / dead-letter behavior in the queue package.
rg -n "Ack|Nack|XAck|XClaim|XAUTOCLAIM|XPENDING|dead.?letter|DLQ|reclaim|pending" internal/queue -SRepository: CeruleanFlow/Cerulean
Length of output: 4973
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the queue interface and all call sites of DequeueBatch/Ack/Nack.
sed -n '1,220p' internal/queue/queue.go
rg -n "DequeueBatch\\(|\\.Ack\\(|\\.Nack\\(" -S .
# Inspect Message definition for context.
rg -n "type Message struct|type Job struct" -S internal/queue
sed -n '1,220p' internal/queue/*.go 2>/dev/null | sed -n '/type Message struct/,/type /p'Repository: CeruleanFlow/Cerulean
Length of output: 939
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Search the whole repo for any Redis stream recovery / dead-letter handling.
rg -n "XAUTOCLAIM|XCLAIM|XPENDING|XDEL|dead.?letter|DLQ|reclaim|retry.*stream|pending forever" -S .
# Search for queue usage to understand whether callers can ever see malformed messages.
rg -n "\.DequeueBatch\(|queue\.Queue|NewRedisStreamQueue|Ack\(|Nack\(" -S .Repository: CeruleanFlow/Cerulean
Length of output: 159
Malformed stream entries need an explicit escape path
DequeueBatch drops messages with a missing payload or bad JSON before returning them, and this queue has no reclaim/dead-letter path. Those entries stay in the consumer PEL and can’t be acked or nacked, so they remain stuck until handled explicitly. Consider acking-and-logging or routing them to a dead-letter flow instead of silently skipping them.
🤖 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 128 - 149, DequeueBatch in
redis_stream.go is silently skipping malformed Redis stream entries with missing
payloads or bad JSON, leaving them stuck in the consumer PEL. Update the loop
that builds messages from stream.Messages to explicitly handle these failures by
logging the bad entry and either acking it immediately or routing it through the
queue’s dead-letter path instead of continuing past it. Use the existing
DequeueBatch flow and the redisMsg.ID/job parsing logic to ensure malformed
records cannot be stranded.
There was a problem hiding this comment.
We shall fix these later, we ll merge this PR
There was a problem hiding this comment.
@Haruko386 Understood, thanks for confirming. Merging now with the fix deferred sounds reasonable — the malformed messages issue is an edge case unlikely to block core functionality.
Would you like me to open a follow-up GitHub issue to track this so it doesn't get lost? Just let me know and I'll create it with a link back to this PR and comment for reference.
ヾ(•ω•`)o
| func (q *RedisStreamQueue) Nack(ctx context.Context, msg Message, reason error) error { | ||
| return nil | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Nack is a permanent no-op.
No requeue, retry-with-backoff, or dead-letter handling is implemented; failed jobs simply remain in the PEL forever with no automatic recovery path (e.g. via XClaim/XAutoClaim). Given the Queue interface commits to this contract, callers relying on Nack for retry semantics will silently get none.
If this is intentionally deferred to a later PR in the stack (e.g. the executor/worker layer), consider a TODO comment here to make that explicit.
🤖 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 159 - 161, The RedisStreamQueue
Nack implementation is a permanent no-op, so failed messages are never requeued,
retried, or dead-lettered. Update Nack on RedisStreamQueue to either implement
the intended retry/recovery path (for example via XClaim/XAutoClaim handling in
the worker flow) or, if this is intentionally deferred, add an explicit
TODO/documentation in Nack and the Queue contract so callers know no retry
semantics are provided yet.
There was a problem hiding this comment.
@Haruko386 Understood — deferring the Nack retry/dead-letter implementation to a later PR, consistent with the previous review, sounds reasonable given the scope here.
To keep this visible for future work, consider adding a short // TODO: comment directly above the Nack stub in internal/queue/redis_stream.go noting that requeue/backoff/dead-letter handling (e.g., via XClaim/XAutoClaim) is intentionally deferred.
Would you like me to open a follow-up GitHub issue to track implementing proper Nack semantics, so it doesn't get lost? Just let me know and I'll create it.
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (1)
internal/search/elastic.go (1)
104-119: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAdd focused tests for the new status branches.
Cover 200, 404, and unexpected statuses, including verification that unexpected responses preserve both status and body in the returned error.
🤖 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/search/elastic.go` around lines 104 - 119, 添加针对检查索引状态逻辑的聚焦测试，覆盖 StatusOK、StatusNotFound 和非预期状态分支；使用可控的 HTTP 响应验证 200/404 成功返回，并确认非预期响应的错误同时包含状态码和响应体内容。
🤖 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/api/handler.go`:
- Around line 270-275: Update the ReindexPaper handler to call
h.ingest.StartPaperReindex instead of h.ingest.StartPaperIngest, preserving the
existing context, paper ID, error handling, and accepted response.
In `@internal/pipeline/paper_ingest.go`:
- Around line 22-37: In PaperIngestHandler.Handle, replace the
ProcessPaperReindex call with ProcessPaperIngest and pass the identifiers in the
target method’s required order: ctx, paperID, taskID. Preserve the existing
validation and trimmed values.
In `@internal/pipeline/paper_reindex.go`:
- Around line 23-33: Update the validation errors in the reindex handler: use a
message describing the missing ingest dependency for the h.ingest nil check, a
task ID-specific message for the empty taskID check, and retain a paper
ID-specific message for the empty paperID check. Locate these checks in the
function containing h.ingest, taskID, and paperID validation.
- Around line 22-37: PaperReindexHandler.Handle passes the identifiers to
ProcessPaperReindex in the wrong order. Update the call to match
ProcessPaperReindex’s signature by passing paperID before taskID, while
preserving the existing validation and context argument.
---
Nitpick comments:
In `@internal/search/elastic.go`:
- Around line 104-119: 添加针对检查索引状态逻辑的聚焦测试，覆盖 StatusOK、StatusNotFound
和非预期状态分支；使用可控的 HTTP 响应验证 200/404 成功返回，并确认非预期响应的错误同时包含状态码和响应体内容。
🪄 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: b7044cff-a434-46a8-8c4b-7c614d12aef4
📒 Files selected for processing (16)
.github/workflows/ci.ymlcmd/server/main.gocmd/worker/main.gointernal/api/handler.gointernal/config/config.gointernal/executor/executor.gointernal/executor/registry.gointernal/executor/worker.gointernal/ingest/service.gointernal/pipeline/log_job.gointernal/pipeline/paper_ingest.gointernal/pipeline/paper_reindex.gointernal/queue/redis_stream.gointernal/search/elastic.goserverworker
✅ Files skipped from review due to trivial changes (1)
- internal/pipeline/log_job.go
| func (h *PaperIngestHandler) Handle(ctx context.Context, job queue.Job) error { | ||
| if h.ingest == nil { | ||
| return fmt.Errorf("ingest service is nil") | ||
| } | ||
|
|
||
| taskID := strings.TrimSpace(job.TaskID) | ||
| paperID := strings.TrimSpace(job.PaperID) | ||
| if taskID == "" { | ||
| return fmt.Errorf("task id is empty") | ||
| } | ||
| if paperID == "" { | ||
| return fmt.Errorf("paper id is empty") | ||
| } | ||
|
|
||
| return h.ingest.ProcessPaperReindex(ctx, taskID, paperID) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Wrong method called with swapped arguments — breaks queued paper ingest entirely.
Two compounding bugs on line 36:
ProcessPaperReindexis called instead ofProcessPaperIngest, sopaper_ingestjobs never runrunPDFTextIngest— they attempt a chunk reindex on a paper that has no chunks yet.- Even ignoring (1), the args are passed as
(ctx, taskID, paperID)while the target method's signature is(ctx context.Context, paperID, taskID string)— so the values land in swapped fields, causings.tasks.Get(ctx, taskID)inside the callee to look up a task by the paper's ID and fail with "task not found" for every job.
This makes the queued ingest pipeline non-functional for all paper_ingest jobs.
🐛 Proposed fix
- return h.ingest.ProcessPaperReindex(ctx, taskID, paperID)
+ return h.ingest.ProcessPaperIngest(ctx, paperID, taskID)📝 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.
| func (h *PaperIngestHandler) Handle(ctx context.Context, job queue.Job) error { | |
| if h.ingest == nil { | |
| return fmt.Errorf("ingest service is nil") | |
| } | |
| taskID := strings.TrimSpace(job.TaskID) | |
| paperID := strings.TrimSpace(job.PaperID) | |
| if taskID == "" { | |
| return fmt.Errorf("task id is empty") | |
| } | |
| if paperID == "" { | |
| return fmt.Errorf("paper id is empty") | |
| } | |
| return h.ingest.ProcessPaperReindex(ctx, taskID, paperID) | |
| } | |
| func (h *PaperIngestHandler) Handle(ctx context.Context, job queue.Job) error { | |
| if h.ingest == nil { | |
| return fmt.Errorf("ingest service is nil") | |
| } | |
| taskID := strings.TrimSpace(job.TaskID) | |
| paperID := strings.TrimSpace(job.PaperID) | |
| if taskID == "" { | |
| return fmt.Errorf("task id is empty") | |
| } | |
| if paperID == "" { | |
| return fmt.Errorf("paper id is empty") | |
| } | |
| 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/pipeline/paper_ingest.go` around lines 22 - 37, In
PaperIngestHandler.Handle, replace the ProcessPaperReindex call with
ProcessPaperIngest and pass the identifiers in the target method’s required
order: ctx, paperID, taskID. Preserve the existing validation and trimmed
values.
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
internal/executor/worker.go (1)
112-166: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueOptional: reuse a single dispatcher instead of allocating one per batch.
handleBatchbuilds a newcelestialdispatcher (and its worker goroutines) on every dequeue cycle. Under steady load this recreates the pool repeatedly. Consider constructing the dispatcher once inNewWorkerand reusing it across batches.🤖 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 112 - 166, Optionally avoid recreating the celestial worker pool for every batch by constructing the dispatcher once in NewWorker and storing it on Worker. Update handleBatch to reuse that dispatcher while preserving the existing configuration and result handling, and ensure its lifecycle is properly managed during worker shutdown.cmd/worker/main.go (1)
132-155: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winFix the misleading search-driver error and drop the commented-out branch.
Two issues:
- An empty
SearchDriverfalls through todefaultand errors out (unlikebuildObjectStorage, which treats""aslocal), yet the error message advertiseslocalas supported.- Lines 134-135 are dead commented-out code.
Either restore a real default/local branch or correct the message to list only the actually-supported drivers.
♻️ Suggested cleanup
func buildSearchBackend(cfg config.Config) (search.Backend, error) { switch strings.ToLower(cfg.SearchDriver) { - //case "", "local": - // return search.NewLocalBackend(), nil - case "elastic", "elasticsearch", "es": backend, err := search.NewElasticBackend(context.Background(), search.ElasticConfig{ URL: cfg.ElasticURL, Index: cfg.ElasticIndex, Username: cfg.ElasticUsername, Password: cfg.ElasticPassword, }) if err != nil { return nil, err } if backend == nil { return nil, fmt.Errorf("elastic backend constructor returned nil") } return backend, nil default: - return nil, fmt.Errorf("unsupported CERULEAN_SEARCH_DRIVER=%q; supported: local, elastic", cfg.SearchDriver) + return nil, fmt.Errorf("unsupported CERULEAN_SEARCH_DRIVER=%q; supported: elastic", cfg.SearchDriver) } }🤖 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 132 - 155, Remove the dead commented-out local branch in buildSearchBackend and make its behavior and error message consistent: either implement a real empty/local case or, if local is unsupported, remove “local” from the advertised supported drivers and keep the default error accurate.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@cmd/worker/main.go`:
- Line 63: Replace the process-local task.NewMemoryManager() in the worker with
a shared Redis- or database-backed task store, and configure cmd/server/main.go
to use the same backend and storage settings so worker updates are visible
through the API’s /tasks/:id endpoints.
In `@internal/executor/worker.go`:
- Around line 199-207: Failed jobs remain pending because Queue.Nack is
currently a no-op. Implement the failure handling flow in the queue
implementation, including retry/reclaim or dead-letter behavior and
removal/acknowledgment of the pending message, then update the worker’s Nack
path to use it consistently; reference the Nack method in
internal/queue/redis_stream.go and the worker error-handling block.
---
Nitpick comments:
In `@cmd/worker/main.go`:
- Around line 132-155: Remove the dead commented-out local branch in
buildSearchBackend and make its behavior and error message consistent: either
implement a real empty/local case or, if local is unsupported, remove “local”
from the advertised supported drivers and keep the default error accurate.
In `@internal/executor/worker.go`:
- Around line 112-166: Optionally avoid recreating the celestial worker pool for
every batch by constructing the dispatcher once in NewWorker and storing it on
Worker. Update handleBatch to reuse that dispatcher while preserving the
existing configuration and result handling, and ensure its lifecycle is properly
managed during worker shutdown.
🪄 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: d99dc8d6-d11c-4fe4-94f8-f4f93b719faa
⛔ Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
📒 Files selected for processing (3)
cmd/worker/main.gogo.modinternal/executor/worker.go
🚧 Files skipped from review as they are similar to previous changes (1)
- go.mod
| log.Fatalf("create search backend: %v", err) | ||
| } | ||
|
|
||
| taskManager := task.NewMemoryManager() |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm server also uses an in-memory manager and whether task state is persisted anywhere shared.
rg -nP --type=go 'NewMemoryManager|task\.Manager|tasks\s' cmd internal/task internal/api internal/ingest -C2Repository: CeruleanFlow/Cerulean
Length of output: 2339
🏁 Script executed:
#!/bin/bash
sed -n '1,220p' internal/task/task.go
printf '\n---\n'
sed -n '1,220p' internal/api/handler.go
printf '\n---\n'
sed -n '1,180p' cmd/server/main.go
printf '\n---\n'
sed -n '1,180p' cmd/worker/main.goRepository: CeruleanFlow/Cerulean
Length of output: 14815
Use a shared task store
task.NewMemoryManager() is process-local, and cmd/server/main.go creates its own separate in-memory manager, so task updates recorded by the worker won’t be visible to the API’s /tasks/:id endpoints. Back this with Redis or the DB instead.
🤖 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` at line 63, Replace the process-local
task.NewMemoryManager() in the worker with a shared Redis- or database-backed
task store, and configure cmd/server/main.go to use the same backend and storage
settings so worker updates are visible through the API’s /tasks/:id endpoints.
| nackCtx, cancel := context.WithTimeout(context.Background(), w.jobTimeout) | ||
| defer cancel() | ||
|
|
||
| if nackErr := w.queue.Nack(nackCtx, msg, err); nackErr != nil { | ||
| log.Printf("nack job failed: redis_id=%s task_id=%s err=%v", msg.RedisID, msg.Job.TaskID, nackErr) | ||
| } | ||
|
|
||
| return result, err | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect Nack and any pending-entry reclaim/retry logic in the redis stream queue.
rg -nP --type=go 'func .*Nack|XAutoClaim|XClaim|XPending|dead.?letter' internal/queue -C3Repository: CeruleanFlow/Cerulean
Length of output: 575
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== queue file outline =="
ast-grep outline internal/queue/redis_stream.go --view expanded || true
echo
echo "== worker file outline =="
ast-grep outline internal/executor/worker.go --view expanded || true
echo
echo "== search for reclaim/retry/pending handling across repo =="
rg -n --type=go 'XAutoClaim|XClaim|XPending|Pending|claim.*pending|dead.?letter|requeue|retry' internal -C 2 || true
echo
echo "== relevant queue methods =="
sed -n '1,260p' internal/queue/redis_stream.goRepository: CeruleanFlow/Cerulean
Length of output: 5587
Failed jobs are stuck in the pending list
internal/queue/redis_stream.go makes Nack a no-op, so errors in internal/executor/worker.go leave messages pending forever with no retry, reclaim, or dead-letter path. Implement the intended failure flow before relying on Nack here.
🤖 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 199 - 207, Failed jobs remain
pending because Queue.Nack is currently a no-op. Implement the failure handling
flow in the queue implementation, including retry/reclaim or dead-letter
behavior and removal/acknowledgment of the pending message, then update the
worker’s Nack path to use it consistently; reference the Nack method in
internal/queue/redis_stream.go and the worker error-handling block.
| messages := make([]Message, 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, | ||
| }) | ||
| } | ||
| } | ||
| return messages, nil | ||
| } |
There was a problem hiding this comment.
We shall fix these later, we ll merge this PR
| func (q *RedisStreamQueue) Nack(ctx context.Context, msg Message, reason error) error { | ||
| return nil | ||
| } |
1 participant
Summary by CodeRabbit