From 2d0c45a52bc12a6a1a1d3715d99d354ccfe528a3 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Wed, 6 May 2026 08:20:29 +0800 Subject: [PATCH] feat(jobs/runners): job_reaper (lease timeout) + job_purge (retention) --- internal/jobs/runners/job_purge.go | 40 +++++++ internal/jobs/runners/job_reaper.go | 43 ++++++++ internal/jobs/runners/reaper_purge_test.go | 121 +++++++++++++++++++++ 3 files changed, 204 insertions(+) create mode 100644 internal/jobs/runners/job_purge.go create mode 100644 internal/jobs/runners/job_reaper.go create mode 100644 internal/jobs/runners/reaper_purge_test.go diff --git a/internal/jobs/runners/job_purge.go b/internal/jobs/runners/job_purge.go new file mode 100644 index 0000000..ce6dc4d --- /dev/null +++ b/internal/jobs/runners/job_purge.go @@ -0,0 +1,40 @@ +package runners + +import ( + "context" + "log/slog" + "time" +) + +// JobsPurgeRepo is the minimal repo surface for purge. +type JobsPurgeRepo interface { + PurgeCompleted(ctx context.Context, completedBefore time.Time) (int64, error) +} + +// JobPurge deletes completed/dead jobs older than retention. Periodic; idempotent. +type JobPurge struct { + repo JobsPurgeRepo + retention time.Duration + log *slog.Logger +} + +// NewJobPurge builds the purger. +func NewJobPurge(repo JobsPurgeRepo, retention time.Duration, log *slog.Logger) *JobPurge { + if log == nil { + log = slog.Default() + } + return &JobPurge{repo: repo, retention: retention, log: log} +} + +// Run purges once. +func (p *JobPurge) Run(ctx context.Context) { + cutoff := time.Now().Add(-p.retention) + rows, err := p.repo.PurgeCompleted(ctx, cutoff) + if err != nil { + p.log.Error("job_purge.purge", "err", err.Error()) + return + } + if rows > 0 { + p.log.Info("job_purge.purged", "count", rows) + } +} diff --git a/internal/jobs/runners/job_reaper.go b/internal/jobs/runners/job_reaper.go new file mode 100644 index 0000000..343241b --- /dev/null +++ b/internal/jobs/runners/job_reaper.go @@ -0,0 +1,43 @@ +package runners + +import ( + "context" + "log/slog" + "time" + + "github.com/google/uuid" +) + +// JobsReapRepo is the minimal jobs repository surface for the reaper. +type JobsReapRepo interface { + ReapStuckRunningJobs(ctx context.Context, leasedBefore time.Time) ([]uuid.UUID, error) +} + +// JobReaper resets jobs.running rows whose lease has exceeded the timeout +// (worker probably crashed). Periodic; idempotent; no retry needed. +type JobReaper struct { + repo JobsReapRepo + leaseTimeout time.Duration + log *slog.Logger +} + +// NewJobReaper builds the reaper. +func NewJobReaper(repo JobsReapRepo, leaseTimeout time.Duration, log *slog.Logger) *JobReaper { + if log == nil { + log = slog.Default() + } + return &JobReaper{repo: repo, leaseTimeout: leaseTimeout, log: log} +} + +// Run reaps stuck rows once. +func (r *JobReaper) Run(ctx context.Context) { + cutoff := time.Now().Add(-r.leaseTimeout) + ids, err := r.repo.ReapStuckRunningJobs(ctx, cutoff) + if err != nil { + r.log.Error("job_reaper.reap", "err", err.Error()) + return + } + if len(ids) > 0 { + r.log.Warn("job_reaper.reaped", "count", len(ids), "ids", ids) + } +} diff --git a/internal/jobs/runners/reaper_purge_test.go b/internal/jobs/runners/reaper_purge_test.go new file mode 100644 index 0000000..a8f424a --- /dev/null +++ b/internal/jobs/runners/reaper_purge_test.go @@ -0,0 +1,121 @@ +package runners_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/yan1h/agent-coding-workflow/internal/jobs/runners" +) + +type fakeJobsReapRepo struct { + reapedIDs []uuid.UUID + reapErr error + capturedCutoff time.Time + called bool +} + +func (f *fakeJobsReapRepo) ReapStuckRunningJobs(_ context.Context, leasedBefore time.Time) ([]uuid.UUID, error) { + f.called = true + f.capturedCutoff = leasedBefore + if f.reapErr != nil { + return nil, f.reapErr + } + return f.reapedIDs, nil +} + +type fakeJobsPurgeRepo struct { + purged int64 + purgeErr error + capturedCutoff time.Time + called bool +} + +func (f *fakeJobsPurgeRepo) PurgeCompleted(_ context.Context, completedBefore time.Time) (int64, error) { + f.called = true + f.capturedCutoff = completedBefore + if f.purgeErr != nil { + return 0, f.purgeErr + } + return f.purged, nil +} + +func TestJobReaper_RunCallsReapWithCorrectCutoff(t *testing.T) { + t.Parallel() + repo := &fakeJobsReapRepo{reapedIDs: []uuid.UUID{uuid.New(), uuid.New()}} + timeout := 5 * time.Minute + r := runners.NewJobReaper(repo, timeout, discardLogger()) + before := time.Now() + r.Run(context.Background()) + after := time.Now() + require.True(t, repo.called) + expectedFloor := before.Add(-timeout).Add(-time.Second) // some slack + expectedCeil := after.Add(-timeout).Add(time.Second) + assert.True(t, !repo.capturedCutoff.Before(expectedFloor) && !repo.capturedCutoff.After(expectedCeil), + "cutoff %s outside [%s, %s]", repo.capturedCutoff, expectedFloor, expectedCeil) +} + +func TestJobReaper_RunErrorDoesNotPanic(t *testing.T) { + t.Parallel() + repo := &fakeJobsReapRepo{reapErr: errors.New("db down")} + r := runners.NewJobReaper(repo, 5*time.Minute, discardLogger()) + require.NotPanics(t, func() { r.Run(context.Background()) }) + assert.True(t, repo.called) +} + +func TestJobReaper_NoStuckJobsIsNoOp(t *testing.T) { + t.Parallel() + repo := &fakeJobsReapRepo{} // empty reapedIDs + r := runners.NewJobReaper(repo, 5*time.Minute, discardLogger()) + require.NotPanics(t, func() { r.Run(context.Background()) }) + assert.True(t, repo.called) +} + +func TestJobPurge_RunCallsPurgeWithCorrectCutoff(t *testing.T) { + t.Parallel() + repo := &fakeJobsPurgeRepo{purged: 17} + retention := 7 * 24 * time.Hour + p := runners.NewJobPurge(repo, retention, discardLogger()) + before := time.Now() + p.Run(context.Background()) + after := time.Now() + require.True(t, repo.called) + expectedFloor := before.Add(-retention).Add(-time.Second) + expectedCeil := after.Add(-retention).Add(time.Second) + assert.True(t, !repo.capturedCutoff.Before(expectedFloor) && !repo.capturedCutoff.After(expectedCeil)) +} + +func TestJobPurge_RunErrorDoesNotPanic(t *testing.T) { + t.Parallel() + repo := &fakeJobsPurgeRepo{purgeErr: errors.New("db error")} + p := runners.NewJobPurge(repo, 7*24*time.Hour, discardLogger()) + require.NotPanics(t, func() { p.Run(context.Background()) }) + assert.True(t, repo.called) +} + +func TestJobPurge_ZeroRowsIsNoOp(t *testing.T) { + t.Parallel() + repo := &fakeJobsPurgeRepo{purged: 0} + p := runners.NewJobPurge(repo, 7*24*time.Hour, discardLogger()) + require.NotPanics(t, func() { p.Run(context.Background()) }) + assert.True(t, repo.called) +} + +func TestJobReaper_NilLogFallsBackToDefault(t *testing.T) { + t.Parallel() + r := runners.NewJobReaper(&fakeJobsReapRepo{}, time.Minute, nil) + require.NotNil(t, r) + require.NotPanics(t, func() { r.Run(context.Background()) }) +} + +func TestJobPurge_NilLogFallsBackToDefault(t *testing.T) { + t.Parallel() + p := runners.NewJobPurge(&fakeJobsPurgeRepo{}, time.Hour, nil) + require.NotNil(t, p) + require.NotPanics(t, func() { p.Run(context.Background()) }) +}