Skip to content

Feature: Add recover pending for redis stream_queue #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Haruko386 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Feature: Add recover pending for redis stream_queue #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
31c602c
Feature: Add recover pending for redis stream_queue
Jul 11, 2026
30c7111
Merge branch 'main' of https://github.com/Haruko386/Cerulean
Jul 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion cmd/worker/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,19 +89,25 @@ func main() {
BlockMillis: 5000,
JobTimeout: 30 * time.Minute,
Concurrency: cfg.WorkerConcurrency,

ClaimInterval: cfg.WorkerClaimInterval,
ClaimMinIdle: cfg.WorkerClaimMinIdle,
ClaimBatchSize: cfg.WorkerClaimBatchSize,
})
if err != nil {
log.Fatalf("create executor worker: %v", err)
}

log.Printf(
"Cerulean worker started: redis=%s stream=%s group=%s consumer=%s batch_size=%d concurrency=%d",
"Cerulean worker started: redis=%s stream=%s group=%s consumer=%s batch_size=%d concurrency=%d job_timeout=%s claim_min_idle=%s",
cfg.RedisAddr,
cfg.QueueStream,
cfg.QueueGroup,
cfg.QueueConsumer,
cfg.WorkerBatchSize,
cfg.WorkerConcurrency,
cfg.WorkerJobTimeout,
cfg.WorkerClaimMinIdle,
)

if err := worker.Run(ctx); err != nil && ctx.Err() == nil {
Expand Down
24 changes: 24 additions & 0 deletions internal/config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package config
import (
"os"
"strconv"
"time"

"github.com/joho/godotenv"
_ "github.com/joho/godotenv"
Expand Down Expand Up @@ -48,6 +49,11 @@ type Config struct {

WorkerConcurrency int
WorkerBatchSize int

WorkerJobTimeout time.Duration
WorkerClaimInterval time.Duration
WorkerClaimMinIdle time.Duration
WorkerClaimBatchSize int
}

func Load() Config {
Expand Down Expand Up @@ -95,6 +101,11 @@ func Load() Config {

WorkerBatchSize: envInt("CERULEAN_WORKER_BATCH_SIZE", 4),
WorkerConcurrency: envInt("CERULEAN_WORKER_CONCURRENCY", 16),

WorkerJobTimeout: envDuration("CERULEAN_WORKER_JOB_TIMEOUT", 30*time.Minute),
WorkerClaimInterval: envDuration("CERULEAN_WORKER_CLAIM_INTERVAL", time.Minute),
WorkerClaimMinIdle: envDuration("CERULEAN_WORKER_CLAIM_MIN_IDLE", 35*time.Minute),
WorkerClaimBatchSize: envInt("CERULEAN_WORKER_CLAIM_BATCH_SIZE", 16),
}
}

Expand All @@ -116,3 +127,16 @@ func envInt(key string, fallback int) int {
}
return i
}

func envDuration(key string, fallback time.Duration) time.Duration {
value := os.Getenv(key)
if value == "" {
return fallback
}
parsed, err := time.ParseDuration(value)
if err != nil {
return fallback
}

return parsed
}
3 changes: 3 additions & 0 deletions internal/executor/registry.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,14 @@ type Registry struct {
handlers map[string]Handler
}

// NewRegistry creates an empty job handler registry.
func NewRegistry() *Registry {
return &Registry{
handlers: make(map[string]Handler),
}
}

// Register registers a handler for a job type.
func (r *Registry) Register(jobType string, handler Handler) error {
jobType = strings.TrimSpace(jobType)
if jobType == "" {
Expand All @@ -32,6 +34,7 @@ func (r *Registry) Register(jobType string, handler Handler) error {
return nil
}

// Execute dispatches a job to its registered handler.
func (r *Registry) Execute(ctx context.Context, job queue.Job) error {
jobType := strings.TrimSpace(job.Type)
if jobType == "" {
Expand Down
99 changes: 91 additions & 8 deletions internal/executor/worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@ package executor
import (
"context"
"errors"
"fmt"
"log"
"strings"
"time"

celestial "github.com/Haruko386/Celestial"
Expand All @@ -19,13 +21,22 @@ type Worker struct {
blockMillis int64
jobTimeout time.Duration
concurrency int

claimInterval time.Duration
claimMinIdle time.Duration
claimBatchSize int
claimCursor string
}

type WorkerOptions struct {
BatchSize int
BlockMillis int64
JobTimeout time.Duration
Concurrency int

ClaimInterval time.Duration
ClaimMinIdle time.Duration
ClaimBatchSize int
}

type JobResult struct {
Expand All @@ -35,6 +46,7 @@ type JobResult struct {
PaperID string
}

// NewWorker creates a configured task executor worker.
func NewWorker(queue queue.Queue, registry *Registry, options WorkerOptions) (*Worker, error) {
if queue == nil {
return nil, errors.New("queue is nil")
Expand Down Expand Up @@ -63,31 +75,71 @@ func NewWorker(queue queue.Queue, registry *Registry, options WorkerOptions) (*W
concurrency = 4
}

claimInterval := options.ClaimInterval
if claimInterval == 0 {
claimInterval = time.Minute
}

claimMinIdle := options.ClaimMinIdle
if claimMinIdle == 0 {
claimMinIdle = jobTimeout + 5*time.Minute
}

claimBatchSize := options.ClaimBatchSize
if claimBatchSize == 0 {
claimBatchSize = batchSize
}

if claimMinIdle < jobTimeout {
return nil, errors.New("claim minimum time limit exceeded")
}

return &Worker{
queue: queue,
registry: *registry,
batchSize: batchSize,
blockMillis: blockMillis,
jobTimeout: jobTimeout,
concurrency: concurrency,
queue: queue,
registry: *registry,
batchSize: batchSize,
blockMillis: blockMillis,
jobTimeout: jobTimeout,
concurrency: concurrency,
claimInterval: claimInterval,
claimMinIdle: claimMinIdle,
claimBatchSize: claimBatchSize,
claimCursor: "0-0",
}, nil
}

// Run starts the worker loop and processes queued jobs.
func (w *Worker) Run(ctx context.Context) error {
log.Printf(
"executor worker started: batch_size=%d block_millis=%d job_timeout=%s",
"executor worker started: batch_size=%d concurrency=%d job_timeout=%s claim_interval=%s claim_min_idle=%s",
w.batchSize,
w.blockMillis,
w.concurrency,
w.jobTimeout,
w.claimInterval,
w.claimMinIdle,
)

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()
}
Comment on lines +122 to +141

@coderabbitai coderabbitai Bot Jul 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 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.

Suggested change
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.


messages, err := w.queue.DequeueBatch(ctx, w.batchSize, w.blockMillis)
if err != nil {
if ctx.Err() != nil {
Expand All @@ -109,6 +161,7 @@ func (w *Worker) Run(ctx context.Context) error {
}
}

// handleBatch processes a batch of messages concurrently.
func (w *Worker) handleBatch(ctx context.Context, messages []queue.Message) error {
if len(messages) == 0 {
return nil
Expand Down Expand Up @@ -165,6 +218,7 @@ func (w *Worker) handleBatch(ctx context.Context, messages []queue.Message) erro
return nil
}

// handleMessage executes and acknowledges a single queue message.
func (w *Worker) handleMessage(ctx context.Context, msg queue.Message) (JobResult, error) {
result := JobResult{
RedisID: msg.RedisID,
Expand Down Expand Up @@ -222,3 +276,32 @@ func (w *Worker) handleMessage(ctx context.Context, msg queue.Message) (JobResul
)
return result, nil
}

// recoverPending claims and processes idle pending messages.
func (w *Worker) recoverPending(ctx context.Context) error {
messages, nextStart, err := w.queue.ClaimPending(ctx, w.claimCursor, w.claimMinIdle, w.claimBatchSize)
if err != nil {
return err
}

w.claimCursor = nextStart
if strings.TrimSpace(nextStart) == "" {
w.claimCursor = "0-0"
}

if len(messages) == 0 {
return nil
}

log.Printf(
"claimed pending jobs: count=%d next_start=%s min_idle=%s",
len(messages),
w.claimCursor,
w.claimMinIdle,
)

if err := w.handleBatch(ctx, messages); err != nil {
return fmt.Errorf("handle claimed pending batch: %w", err)
}
return nil
}
17 changes: 15 additions & 2 deletions internal/ingest/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ func NewService(
}
}

// StartPaperIngest creates and enqueues a paper ingestion task.
func (s *Service) StartPaperIngest(ctx context.Context, paperID string) (task.Task, error) {
paperID = strings.TrimSpace(paperID)
if paperID == "" {
Expand Down Expand Up @@ -105,6 +106,7 @@ func (s *Service) StartPaperIngest(ctx context.Context, paperID string) (task.Ta
return job, nil
}

// StartPaperReindex creates and enqueues a paper reindex task.
func (s *Service) StartPaperReindex(ctx context.Context, paperID string) (task.Task, error) {
paperID = strings.TrimSpace(paperID)
if paperID == "" {
Expand Down Expand Up @@ -153,6 +155,7 @@ func (s *Service) StartPaperReindex(ctx context.Context, paperID string) (task.T
return job, nil
}

// runPDFTextIngest parses a PDF and stores its generated artifacts and chunks.
func (s *Service) runPDFTextIngest(ctx context.Context, job task.Task, paper domain.Paper) error {
// precheck
if s.parser == nil {
Expand Down Expand Up @@ -236,7 +239,7 @@ func (s *Service) runPDFTextIngest(ctx context.Context, job task.Task, paper dom
return nil
}

// fail set failed status for paper and task
// fail marks both the paper and task as failed.
func (s *Service) fail(ctx context.Context, job task.Task, paper domain.Paper, err error) {
opCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
Expand All @@ -254,6 +257,7 @@ func (s *Service) fail(ctx context.Context, job task.Task, paper domain.Paper, e
_ = s.tasks.Update(opCtx, job)
}

// failTaskOnly marks only the task as failed.
func (s *Service) failTaskOnly(ctx context.Context, job task.Task, err error) {
opCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
Expand All @@ -266,7 +270,7 @@ func (s *Service) failTaskOnly(ctx context.Context, job task.Task, err error) {
_ = s.tasks.Update(opCtx, job)
}

// downloadOriginalPDF download original PDF to tmp
// downloadOriginalPDF downloads the original PDF to a temporary local file.
func (s *Service) downloadOriginalPDF(ctx context.Context, paper domain.Paper) (string, func(), error) {
if s.store == nil {
return "", nil, fmt.Errorf("object storage is not initialized")
Expand Down Expand Up @@ -323,6 +327,7 @@ func parsedMarkdown(paper domain.Paper, doc docparser.Document) string {
return b.String()
}

// ReindexPaper rebuilds a paper's Elasticsearch index from MySQL chunks.
func (s *Service) ReindexPaper(ctx context.Context, paperID string) error {
if s.search == nil {
return fmt.Errorf("search backend is not initialized")
Expand Down Expand Up @@ -355,6 +360,7 @@ func (s *Service) ReindexPaper(ctx context.Context, paperID string) error {
return nil
}

// ProcessPaperIngest executes a queued paper ingestion task.
func (s *Service) ProcessPaperIngest(ctx context.Context, paperID, taskID string) error {
taskID = strings.TrimSpace(taskID)
paperID = strings.TrimSpace(paperID)
Expand All @@ -370,6 +376,9 @@ func (s *Service) ProcessPaperIngest(ctx context.Context, paperID, taskID string
if !ok {
return fmt.Errorf("task %s not found", taskID)
}
if job.Status == task.Succeeded {
return nil
}
Comment on lines +379 to +381

@coderabbitai coderabbitai Bot Jul 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ 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.


paper, err := s.papers.Get(ctx, paperID)
if err != nil {
Expand All @@ -383,6 +392,7 @@ func (s *Service) ProcessPaperIngest(ctx context.Context, paperID, taskID string
return nil
}

// ProcessPaperReindex executes a queued paper reindex task.
func (s *Service) ProcessPaperReindex(ctx context.Context, paperID, taskID string) error {
taskID = strings.TrimSpace(taskID)
paperID = strings.TrimSpace(paperID)
Expand All @@ -398,6 +408,9 @@ func (s *Service) ProcessPaperReindex(ctx context.Context, paperID, taskID strin
if !ok {
return fmt.Errorf("task %s not found", taskID)
}
if job.Status == task.Succeeded {
return nil
}

now := time.Now()
job.Status = task.Running
Expand Down
4 changes: 3 additions & 1 deletion internal/parser/chunker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"github.com/CeruleanFlow/cerulean/internal/domain"
)

// BuildChunks converts parsed document pages into searchable chunks.
func BuildChunks(paper domain.Paper, artifactKey string, doc Document, maxRunes, overlap int) []domain.Chunk {
if maxRunes <= 0 {
maxRunes = 1200
Expand Down Expand Up @@ -56,7 +57,7 @@ func BuildChunks(paper domain.Paper, artifactKey string, doc Document, maxRunes,
return chunks
}

// splitTextByRunes split the text by runes
// splitTextByRunes splits text into overlapping UTF-8-safe chunks.
func splitTextByRunes(text string, maxRunes int, overlap int) []string {
runes := []rune(text)
if len(runes) == 0 {
Expand Down Expand Up @@ -90,6 +91,7 @@ func splitTextByRunes(text string, maxRunes int, overlap int) []string {
return result
}

// chooseChunkEnd selects a natural sentence boundary for a chunk.
func chooseChunkEnd(runes []rune, start int, end int) int {
minEnd := start + int(float64(end-start)*0.7)

Expand Down
1 change: 1 addition & 0 deletions internal/parser/pdf_text.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ func NewPDFTextParser() *PDFTextParser {
return &PDFTextParser{}
}

// ParseFile extracts page-level text from a PDF file.
func (p *PDFTextParser) ParseFile(ctx context.Context, path string) (Document, error) {
f, reader, err := pdf.Open(path)
if err != nil {
Expand Down
Loading
Loading