diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go new file mode 100644 index 0000000..be6e587 --- /dev/null +++ b/internal/jobs/worker.go @@ -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] +} diff --git a/internal/jobs/worker_test.go b/internal/jobs/worker_test.go new file mode 100644 index 0000000..de19863 --- /dev/null +++ b/internal/jobs/worker_test.go @@ -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() +}