You've already forked agentic-coding-workflow
44 lines
1.1 KiB
Go
44 lines
1.1 KiB
Go
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)
|
|
}
|
|
}
|