You've already forked agentic-coding-workflow
feat(jobs): worker pool with retry/dead-letter/panic-recovery
This commit is contained in:
@@ -0,0 +1,148 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/clock"
|
||||
)
|
||||
|
||||
// WorkerOptions configures a single Worker. ID identifies the worker in
|
||||
// jobs.leased_by. Backoff is the per-attempt retry delay sequence; if attempts
|
||||
// exceeds len(Backoff), the last entry is used.
|
||||
type WorkerOptions struct {
|
||||
ID string
|
||||
PollInterval time.Duration
|
||||
Backoff []time.Duration
|
||||
Clock clock.Clock
|
||||
Log *slog.Logger
|
||||
}
|
||||
|
||||
// Worker pulls leased jobs from the repository, dispatches them to handlers,
|
||||
// and writes back the appropriate completion / retry / dead-letter state.
|
||||
type Worker struct {
|
||||
repo Repository
|
||||
handlers map[JobType]Handler
|
||||
opts WorkerOptions
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// NewWorker builds a Worker. The caller registers handlers by JobType.
|
||||
func NewWorker(repo Repository, handlers map[JobType]Handler, opts WorkerOptions) *Worker {
|
||||
if opts.PollInterval <= 0 {
|
||||
opts.PollInterval = 100 * time.Millisecond
|
||||
}
|
||||
if opts.Clock == nil {
|
||||
opts.Clock = clock.Real()
|
||||
}
|
||||
if opts.Log == nil {
|
||||
opts.Log = slog.Default()
|
||||
}
|
||||
if len(opts.Backoff) == 0 {
|
||||
opts.Backoff = []time.Duration{30 * time.Second}
|
||||
}
|
||||
return &Worker{repo: repo, handlers: handlers, opts: opts}
|
||||
}
|
||||
|
||||
// Run loops until ctx is canceled, polling for and processing jobs.
|
||||
func (w *Worker) Run(ctx context.Context) {
|
||||
w.wg.Add(1)
|
||||
defer w.wg.Done()
|
||||
t := time.NewTicker(w.opts.PollInterval)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
w.pollOnce(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Wait blocks until the goroutine started by Run returns.
|
||||
func (w *Worker) Wait() { w.wg.Wait() }
|
||||
|
||||
func (w *Worker) pollOnce(ctx context.Context) {
|
||||
job, err := w.repo.Lease(ctx, w.opts.ID)
|
||||
if err != nil {
|
||||
w.opts.Log.Error("job.lease_failed", "err", err.Error())
|
||||
return
|
||||
}
|
||||
if job == nil {
|
||||
return
|
||||
}
|
||||
w.execute(ctx, *job)
|
||||
}
|
||||
|
||||
func (w *Worker) execute(ctx context.Context, job Job) {
|
||||
w.opts.Log.Info("job.start", "id", job.ID, "type", job.Type, "attempt", job.Attempts)
|
||||
start := w.opts.Clock.Now()
|
||||
|
||||
h, ok := w.handlers[job.Type]
|
||||
if !ok {
|
||||
_ = w.repo.DeadLetter(ctx, job.ID, fmt.Sprintf("unknown job type: %s", job.Type))
|
||||
w.opts.Log.Error("job.dead_letter", "id", job.ID, "type", job.Type, "reason", "unknown_type")
|
||||
return
|
||||
}
|
||||
|
||||
err := w.runHandler(ctx, h, job)
|
||||
if err == nil {
|
||||
_ = w.repo.Complete(ctx, job.ID)
|
||||
w.opts.Log.Info("job.complete",
|
||||
"id", job.ID, "type", job.Type, "attempt", job.Attempts,
|
||||
"duration_ms", w.opts.Clock.Now().Sub(start).Milliseconds())
|
||||
return
|
||||
}
|
||||
|
||||
// Permanent or attempts exhausted -> dead-letter
|
||||
if IsPermanent(err) || job.Attempts >= job.MaxAttempts {
|
||||
msg := truncate(err.Error(), 2000)
|
||||
_ = w.repo.DeadLetter(ctx, job.ID, msg)
|
||||
w.opts.Log.Error("job.dead_letter",
|
||||
"id", job.ID, "type", job.Type, "attempts", job.Attempts, "err", msg)
|
||||
return
|
||||
}
|
||||
|
||||
// Retryable
|
||||
delay := backoffFor(w.opts.Backoff, int(job.Attempts))
|
||||
next := w.opts.Clock.Now().Add(delay)
|
||||
msg := truncate(err.Error(), 2000)
|
||||
_ = w.repo.FailWithRetry(ctx, job.ID, next, msg)
|
||||
w.opts.Log.Warn("job.fail_retry",
|
||||
"id", job.ID, "type", job.Type, "attempt", job.Attempts,
|
||||
"err", msg, "next_at", next.Format(time.RFC3339))
|
||||
}
|
||||
|
||||
// runHandler wraps Handle with panic recovery, converting panics into retryable errors.
|
||||
func (w *Worker) runHandler(ctx context.Context, h Handler, job Job) (err error) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
err = fmt.Errorf("handler panic: %v", r)
|
||||
}
|
||||
}()
|
||||
return h.Handle(ctx, job)
|
||||
}
|
||||
|
||||
// backoffFor returns the backoff for attempt index (1-based "after first try").
|
||||
// If index exceeds the slice, the last element is used.
|
||||
func backoffFor(seq []time.Duration, attempt int) time.Duration {
|
||||
if attempt <= 0 {
|
||||
return seq[0]
|
||||
}
|
||||
idx := attempt - 1
|
||||
if idx >= len(seq) {
|
||||
idx = len(seq) - 1
|
||||
}
|
||||
return seq[idx]
|
||||
}
|
||||
|
||||
func truncate(s string, n int) string {
|
||||
if len(s) <= n {
|
||||
return s
|
||||
}
|
||||
return s[:n]
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
//go:build integration
|
||||
|
||||
package jobs_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/clock"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/jobs"
|
||||
)
|
||||
|
||||
type recordingHandler struct {
|
||||
typ jobs.JobType
|
||||
calls atomic.Int32
|
||||
wantErr error
|
||||
beforeRet func()
|
||||
}
|
||||
|
||||
func (h *recordingHandler) Type() jobs.JobType { return h.typ }
|
||||
func (h *recordingHandler) Handle(ctx context.Context, job jobs.Job) error {
|
||||
h.calls.Add(1)
|
||||
if h.beforeRet != nil {
|
||||
h.beforeRet()
|
||||
}
|
||||
return h.wantErr
|
||||
}
|
||||
|
||||
func TestWorker_HandlesAndCompletes(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, _ := setupRepo(t)
|
||||
_, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
|
||||
require.NoError(t, err)
|
||||
|
||||
h := &recordingHandler{typ: jobs.TypeWebhookDeliver}
|
||||
w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
|
||||
jobs.WorkerOptions{
|
||||
ID: "test#1",
|
||||
PollInterval: 5 * time.Millisecond,
|
||||
Backoff: []time.Duration{30 * time.Second},
|
||||
Clock: clock.Real(),
|
||||
Log: discardLogger(),
|
||||
})
|
||||
cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
|
||||
defer cancel()
|
||||
go w.Run(cctx)
|
||||
require.Eventually(t, func() bool { return h.calls.Load() >= 1 }, 200*time.Millisecond, 5*time.Millisecond)
|
||||
cancel()
|
||||
w.Wait()
|
||||
}
|
||||
|
||||
func TestWorker_RetryableErrorReschedules(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, _ := setupRepo(t)
|
||||
enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
|
||||
require.NoError(t, err)
|
||||
|
||||
h := &recordingHandler{typ: jobs.TypeWebhookDeliver, wantErr: errors.New("transient")}
|
||||
w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
|
||||
jobs.WorkerOptions{
|
||||
ID: "t#1",
|
||||
PollInterval: 5 * time.Millisecond,
|
||||
Backoff: []time.Duration{30 * time.Second, 2 * time.Minute},
|
||||
Clock: clock.Real(),
|
||||
Log: discardLogger(),
|
||||
})
|
||||
cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
|
||||
defer cancel()
|
||||
go w.Run(cctx)
|
||||
require.Eventually(t, func() bool { return h.calls.Load() >= 1 }, 200*time.Millisecond, 5*time.Millisecond)
|
||||
cancel()
|
||||
w.Wait()
|
||||
|
||||
got, err := repo.Get(ctx, enqueued.ID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, jobs.StatusPending, got.Status)
|
||||
assert.Equal(t, int32(1), got.Attempts)
|
||||
require.NotNil(t, got.LastError)
|
||||
assert.Equal(t, "transient", *got.LastError)
|
||||
}
|
||||
|
||||
func TestWorker_PermanentErrorDeadLetters(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, _ := setupRepo(t)
|
||||
enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
|
||||
require.NoError(t, err)
|
||||
h := &recordingHandler{typ: jobs.TypeWebhookDeliver, wantErr: jobs.Permanent(errors.New("4xx config error"))}
|
||||
w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
|
||||
jobs.WorkerOptions{ID: "t", PollInterval: 5 * time.Millisecond, Backoff: []time.Duration{time.Second}, Clock: clock.Real(), Log: discardLogger()})
|
||||
cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
|
||||
defer cancel()
|
||||
go w.Run(cctx)
|
||||
require.Eventually(t, func() bool {
|
||||
got, _ := repo.Get(ctx, enqueued.ID)
|
||||
return got != nil && got.Status == jobs.StatusDead
|
||||
}, 300*time.Millisecond, 10*time.Millisecond)
|
||||
cancel()
|
||||
w.Wait()
|
||||
}
|
||||
|
||||
func TestWorker_MaxAttemptsExhaustedDeadLetters(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, _ := setupRepo(t)
|
||||
enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 2) // max=2
|
||||
require.NoError(t, err)
|
||||
h := &recordingHandler{typ: jobs.TypeWebhookDeliver, wantErr: errors.New("transient")}
|
||||
w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
|
||||
jobs.WorkerOptions{ID: "t", PollInterval: 2 * time.Millisecond, Backoff: []time.Duration{time.Millisecond}, Clock: clock.Real(), Log: discardLogger()})
|
||||
cctx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
|
||||
defer cancel()
|
||||
go w.Run(cctx)
|
||||
require.Eventually(t, func() bool {
|
||||
got, _ := repo.Get(ctx, enqueued.ID)
|
||||
return got != nil && got.Status == jobs.StatusDead
|
||||
}, 500*time.Millisecond, 10*time.Millisecond)
|
||||
cancel()
|
||||
w.Wait()
|
||||
}
|
||||
|
||||
func TestWorker_HandlerPanicTreatedAsRetry(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, _ := setupRepo(t)
|
||||
enqueued, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
|
||||
require.NoError(t, err)
|
||||
h := &recordingHandler{typ: jobs.TypeWebhookDeliver, beforeRet: func() { panic("kaboom") }}
|
||||
w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{jobs.TypeWebhookDeliver: h},
|
||||
jobs.WorkerOptions{ID: "t", PollInterval: 5 * time.Millisecond, Backoff: []time.Duration{time.Second}, Clock: clock.Real(), Log: discardLogger()})
|
||||
cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
|
||||
defer cancel()
|
||||
go w.Run(cctx)
|
||||
require.Eventually(t, func() bool {
|
||||
got, _ := repo.Get(ctx, enqueued.ID)
|
||||
return got != nil && got.Status == jobs.StatusPending && got.Attempts >= 1
|
||||
}, 300*time.Millisecond, 10*time.Millisecond)
|
||||
cancel()
|
||||
w.Wait()
|
||||
}
|
||||
|
||||
func TestWorker_UnknownTypeDeadLetters(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, _ := setupRepo(t)
|
||||
enqueued, err := repo.Enqueue(ctx, "unknown.type", json.RawMessage(`{}`), time.Now(), 5)
|
||||
require.NoError(t, err)
|
||||
w := jobs.NewWorker(repo, map[jobs.JobType]jobs.Handler{},
|
||||
jobs.WorkerOptions{ID: "t", PollInterval: 5 * time.Millisecond, Backoff: []time.Duration{time.Second}, Clock: clock.Real(), Log: discardLogger()})
|
||||
cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
|
||||
defer cancel()
|
||||
go w.Run(cctx)
|
||||
require.Eventually(t, func() bool {
|
||||
got, _ := repo.Get(ctx, enqueued.ID)
|
||||
return got != nil && got.Status == jobs.StatusDead
|
||||
}, 200*time.Millisecond, 10*time.Millisecond)
|
||||
cancel()
|
||||
w.Wait()
|
||||
}
|
||||
Reference in New Issue
Block a user