feat(jobs/runners): job_reaper (lease timeout) + job_purge (retention)

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)
+	}
+}

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)
+	}
+}

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()) })
+}