q792602257/agentic-coding-workflow
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(jobs): log repo write errors in worker.execute

Browse Source 
Previously, write-back errors from Complete/DeadLetter/FailWithRetry
were silently dropped via underscore assignment. A transient DB
failure here would leave a job stuck (still 'running' with active
lease) until the reaper resets it, with zero observability into why.

Log each write-back error at Error level so operators can correlate
stuck jobs with DB symptoms. Code quality review of T8.
This commit is contained in:
Jerry Yan
2026-05-05 23:12:43 +08:00
parent 1a83a5b1cf
commit 3aa19c8f7e
1 changed files with 12 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/jobs/worker.go
+12 -4
View File
@@ -84,14 +84,18 @@ func (w *Worker) execute(ctx context.Context, job Job) {
h, ok := w.handlers[job.Type] h, ok := w.handlers[job.Type]
if !ok { if !ok {
_ = w.repo.DeadLetter(ctx, job.ID, fmt.Sprintf("unknown job type: %s", job.Type)) if derr := w.repo.DeadLetter(ctx, job.ID, fmt.Sprintf("unknown job type: %s", job.Type)); derr != nil {
w.opts.Log.Error("job.dead_letter_write_failed", "id", job.ID, "err", derr.Error())
}
w.opts.Log.Error("job.dead_letter", "id", job.ID, "type", job.Type, "reason", "unknown_type") w.opts.Log.Error("job.dead_letter", "id", job.ID, "type", job.Type, "reason", "unknown_type")
return return
} }
err := w.runHandler(ctx, h, job) err := w.runHandler(ctx, h, job)
if err == nil { if err == nil {
_ = w.repo.Complete(ctx, job.ID) if cerr := w.repo.Complete(ctx, job.ID); cerr != nil {
w.opts.Log.Error("job.complete_write_failed", "id", job.ID, "err", cerr.Error())
}
w.opts.Log.Info("job.complete", w.opts.Log.Info("job.complete",
"id", job.ID, "type", job.Type, "attempt", job.Attempts, "id", job.ID, "type", job.Type, "attempt", job.Attempts,
"duration_ms", w.opts.Clock.Now().Sub(start).Milliseconds()) "duration_ms", w.opts.Clock.Now().Sub(start).Milliseconds())
@@ -101,7 +105,9 @@ func (w *Worker) execute(ctx context.Context, job Job) {
// Permanent or attempts exhausted -> dead-letter // Permanent or attempts exhausted -> dead-letter
if IsPermanent(err) || job.Attempts >= job.MaxAttempts { if IsPermanent(err) || job.Attempts >= job.MaxAttempts {
msg := truncate(err.Error(), 2000) msg := truncate(err.Error(), 2000)
_ = w.repo.DeadLetter(ctx, job.ID, msg) if derr := w.repo.DeadLetter(ctx, job.ID, msg); derr != nil {
w.opts.Log.Error("job.dead_letter_write_failed", "id", job.ID, "err", derr.Error())
}
w.opts.Log.Error("job.dead_letter", w.opts.Log.Error("job.dead_letter",
"id", job.ID, "type", job.Type, "attempts", job.Attempts, "err", msg) "id", job.ID, "type", job.Type, "attempts", job.Attempts, "err", msg)
return return
@@ -111,7 +117,9 @@ func (w *Worker) execute(ctx context.Context, job Job) {
delay := backoffFor(w.opts.Backoff, int(job.Attempts)) delay := backoffFor(w.opts.Backoff, int(job.Attempts))
next := w.opts.Clock.Now().Add(delay) next := w.opts.Clock.Now().Add(delay)
msg := truncate(err.Error(), 2000) msg := truncate(err.Error(), 2000)
_ = w.repo.FailWithRetry(ctx, job.ID, next, msg) if ferr := w.repo.FailWithRetry(ctx, job.ID, next, msg); ferr != nil {
w.opts.Log.Error("job.fail_retry_write_failed", "id", job.ID, "err", ferr.Error())
}
w.opts.Log.Warn("job.fail_retry", w.opts.Log.Warn("job.fail_retry",
"id", job.ID, "type", job.Type, "attempt", job.Attempts, "id", job.ID, "type", job.Type, "attempt", job.Attempts,
"err", msg, "next_at", next.Format(time.RFC3339)) "err", msg, "next_at", next.Format(time.RFC3339))