feat: expirar jobs stuck com status timeout

Adiciona mecanismo de expiração periódica para jobs que ficam presos
nos status received ou processing além de um threshold configurável.

- models: novo status StatusTimeout
- store: migration v2 com índice em jobs(status); método ExpireStuck
  que faz UPDATE atômico para os status alvo anteriores ao threshold
- expirer: novo pacote com goroutine periódica que chama ExpireStuck
  e loga WARN quando algum job é expirado
- config: campos JobTimeoutMinutes (default 120) e
  ExpirerIntervalMinutes (default 10), lidos via env vars
  JOB_TIMEOUT_MINUTES e EXPIRER_INTERVAL_MINUTES
- main: instancia e inicia o Expirer no startup
- dashboard: card Timeout com contagem; badge neutral na tabela de jobs
This commit is contained in:
Rodrigo Verdiani 2026-07-07 08:43:33 -03:00
parent 7e5f4cb155
commit 2eead05c34
9 changed files with 159 additions and 39 deletions

View File

@ -12,6 +12,7 @@ import (
"time" "time"
"coda/internal/config" "coda/internal/config"
"coda/internal/expirer"
"coda/internal/jobs" "coda/internal/jobs"
"coda/internal/lidarr" "coda/internal/lidarr"
"coda/internal/logger" "coda/internal/logger"
@ -74,6 +75,13 @@ func main() {
} }
pool.Start(ctx) pool.Start(ctx)
exp := expirer.New(
st.Jobs,
time.Duration(envCfg.JobTimeoutMinutes)*time.Minute,
time.Duration(envCfg.ExpirerIntervalMinutes)*time.Minute,
)
exp.Start(ctx)
r := chi.NewRouter() r := chi.NewRouter()
r.Use(middleware.RequestID) r.Use(middleware.RequestID)
r.Use(middleware.Recoverer) r.Use(middleware.Recoverer)

View File

@ -68,6 +68,8 @@ type Config struct {
NtfyURL string NtfyURL string
NtfyTopic string NtfyTopic string
NtfyToken string NtfyToken string
JobTimeoutMinutes int
ExpirerIntervalMinutes int
} }
func Load() (*Config, error) { func Load() (*Config, error) {
@ -102,6 +104,8 @@ func buildConfig(overrides map[string]string) *Config {
NtfyURL: mergeString(overrides, "NTFY_URL", ""), NtfyURL: mergeString(overrides, "NTFY_URL", ""),
NtfyTopic: mergeString(overrides, "NTFY_TOPIC", ""), NtfyTopic: mergeString(overrides, "NTFY_TOPIC", ""),
NtfyToken: mergeString(overrides, "NTFY_TOKEN", ""), NtfyToken: mergeString(overrides, "NTFY_TOKEN", ""),
JobTimeoutMinutes: mergeInt(overrides, "JOB_TIMEOUT_MINUTES", 120),
ExpirerIntervalMinutes: mergeInt(overrides, "EXPIRER_INTERVAL_MINUTES", 10),
} }
} }

View File

@ -0,0 +1,57 @@
package expirer
import (
"context"
"log/slog"
"time"
"coda/internal/models"
)
type JobExpirer interface {
ExpireStuck(ctx context.Context, statuses []models.JobStatus, olderThan time.Duration) (int, error)
}
type Expirer struct {
jobs JobExpirer
timeout time.Duration
interval time.Duration
}
func New(jobs JobExpirer, timeout, interval time.Duration) *Expirer {
return &Expirer{
jobs: jobs,
timeout: timeout,
interval: interval,
}
}
func (e *Expirer) Start(ctx context.Context) {
go e.run(ctx)
}
func (e *Expirer) run(ctx context.Context) {
ticker := time.NewTicker(e.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
e.expire(ctx)
}
}
}
func (e *Expirer) expire(ctx context.Context) {
statuses := []models.JobStatus{models.StatusReceived, models.StatusProcessing}
n, err := e.jobs.ExpireStuck(ctx, statuses, e.timeout)
if err != nil {
slog.Error("expirer: failed to expire stuck jobs", "error", err)
return
}
if n > 0 {
slog.Warn("expirer: expired stuck jobs", "count", n, "timeout", e.timeout)
}
}

View File

@ -9,6 +9,7 @@ const (
StatusProcessing JobStatus = "processing" StatusProcessing JobStatus = "processing"
StatusDone JobStatus = "done" StatusDone JobStatus = "done"
StatusError JobStatus = "error" StatusError JobStatus = "error"
StatusTimeout JobStatus = "timeout"
) )
type JobSource string type JobSource string

View File

@ -4,6 +4,7 @@ import (
"context" "context"
"database/sql" "database/sql"
"fmt" "fmt"
"strings"
"time" "time"
"coda/internal/models" "coda/internal/models"
@ -145,3 +146,35 @@ func (r *JobRepository) Finish(ctx context.Context, id string, status models.Job
} }
return nil return nil
} }
func (r *JobRepository) ExpireStuck(ctx context.Context, statuses []models.JobStatus, olderThan time.Duration) (int, error) {
if len(statuses) == 0 {
return 0, nil
}
placeholders := make([]string, len(statuses))
args := make([]any, 0, len(statuses)+1)
for i, status := range statuses {
placeholders[i] = "?"
args = append(args, status)
}
args = append(args, time.Now().UTC().Add(-olderThan))
query := fmt.Sprintf(
`UPDATE jobs SET status = ?, finished_at = ?, error_msg = ? WHERE status IN (%s) AND created_at < ?`,
strings.Join(placeholders, ","),
)
finalArgs := []any{models.StatusTimeout, time.Now().UTC(), "job timed out"}
finalArgs = append(finalArgs, args...)
result, err := r.db.ExecContext(ctx, query, finalArgs...)
if err != nil {
return 0, fmt.Errorf("expire stuck jobs: %w", err)
}
rows, err := result.RowsAffected()
if err != nil {
return 0, fmt.Errorf("rows affected: %w", err)
}
return int(rows), nil
}

View File

@ -59,6 +59,12 @@ func migrate(ctx context.Context, db *sql.DB) error {
)`, )`,
}, },
}, },
{
version: 2,
sql: []string{
`CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)`,
},
},
} }
for _, m := range migrations { for _, m := range migrations {

View File

@ -23,6 +23,7 @@ type dashboardData struct {
ProcessingCount int ProcessingCount int
DoneCount int DoneCount int
ErrorCount int ErrorCount int
TimeoutCount int
} }
func (h *Handler) Dashboard(w http.ResponseWriter, r *http.Request) { func (h *Handler) Dashboard(w http.ResponseWriter, r *http.Request) {
@ -45,6 +46,7 @@ func (h *Handler) Dashboard(w http.ResponseWriter, r *http.Request) {
dd.ProcessingCount = h.jobCount(ctx, models.StatusProcessing) dd.ProcessingCount = h.jobCount(ctx, models.StatusProcessing)
dd.DoneCount = h.jobCount(ctx, models.StatusDone) dd.DoneCount = h.jobCount(ctx, models.StatusDone)
dd.ErrorCount = h.jobCount(ctx, models.StatusError) dd.ErrorCount = h.jobCount(ctx, models.StatusError)
dd.TimeoutCount = h.jobCount(ctx, models.StatusTimeout)
h.render.Render(w, "dashboard.html", dd) h.render.Render(w, "dashboard.html", dd)
} }
@ -64,6 +66,7 @@ func (h *Handler) Status(w http.ResponseWriter, r *http.Request) {
ProcessingCount: h.jobCount(ctx, models.StatusProcessing), ProcessingCount: h.jobCount(ctx, models.StatusProcessing),
DoneCount: h.jobCount(ctx, models.StatusDone), DoneCount: h.jobCount(ctx, models.StatusDone),
ErrorCount: h.jobCount(ctx, models.StatusError), ErrorCount: h.jobCount(ctx, models.StatusError),
TimeoutCount: h.jobCount(ctx, models.StatusTimeout),
} }
h.render.Render(w, "status.html", dd) h.render.Render(w, "status.html", dd)

View File

@ -81,6 +81,8 @@ func statusBadge(level string) template.HTML {
return template.HTML(`<span class="badge badge-error">` + level + `</span>`) return template.HTML(`<span class="badge badge-error">` + level + `</span>`)
case "received": case "received":
return template.HTML(`<span class="badge badge-info">` + level + `</span>`) return template.HTML(`<span class="badge badge-info">` + level + `</span>`)
case "timeout":
return template.HTML(`<span class="badge badge-neutral">` + level + `</span>`)
default: default:
return template.HTML(level) return template.HTML(level)
} }

View File

@ -31,7 +31,7 @@
</div> </div>
</div> </div>
<div class="grid grid-cols-2 md:grid-cols-4 gap-4"> <div class="grid grid-cols-2 md:grid-cols-5 gap-4">
<div class="card bg-base-100 shadow"> <div class="card bg-base-100 shadow">
<div class="card-body"> <div class="card-body">
<h2 class="card-title">Received</h2> <h2 class="card-title">Received</h2>
@ -56,6 +56,12 @@
<p class="text-4xl font-bold text-error">{{ .ErrorCount }}</p> <p class="text-4xl font-bold text-error">{{ .ErrorCount }}</p>
</div> </div>
</div> </div>
<div class="card bg-base-100 shadow">
<div class="card-body">
<h2 class="card-title">Timeout</h2>
<p class="text-4xl font-bold text-neutral">{{ .TimeoutCount }}</p>
</div>
</div>
</div> </div>
<div class="card bg-base-100 shadow"> <div class="card bg-base-100 shadow">