From 2eead05c34c650b06f7f119c867fb43a098b3d67 Mon Sep 17 00:00:00 2001 From: Rodrigo Verdiani Date: Tue, 7 Jul 2026 08:43:33 -0300 Subject: [PATCH] feat: expirar jobs stuck com status timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- cmd/server/main.go | 8 ++++ internal/config/config.go | 80 +++++++++++++++++++----------------- internal/expirer/expirer.go | 57 +++++++++++++++++++++++++ internal/models/models.go | 1 + internal/store/jobs.go | 33 +++++++++++++++ internal/store/migrations.go | 6 +++ internal/web/dashboard.go | 3 ++ internal/web/render.go | 2 + web/templates/dashboard.html | 8 +++- 9 files changed, 159 insertions(+), 39 deletions(-) create mode 100644 internal/expirer/expirer.go diff --git a/cmd/server/main.go b/cmd/server/main.go index 98ce038..4fb761d 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -12,6 +12,7 @@ import ( "time" "coda/internal/config" + "coda/internal/expirer" "coda/internal/jobs" "coda/internal/lidarr" "coda/internal/logger" @@ -74,6 +75,13 @@ func main() { } 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.Use(middleware.RequestID) r.Use(middleware.Recoverer) diff --git a/internal/config/config.go b/internal/config/config.go index 445668a..7c0637d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -49,25 +49,27 @@ var editableKeys = map[Key]struct{}{ var timeRegex = regexp.MustCompile(`^\d{2}:\d{2}$`) type Config struct { - Port string - Workers int - FFmpegBin string - FFprobeBin string - LogLevel string - DryRun bool - LidarrURL string - LidarrAPIKey string - WebhookSecret string - MusicPath string - DBPath string - Timezone string - WorkWindowEnabled bool - WorkWindowStart string - WorkWindowEnd string - NtfyEnabled bool - NtfyURL string - NtfyTopic string - NtfyToken string + Port string + Workers int + FFmpegBin string + FFprobeBin string + LogLevel string + DryRun bool + LidarrURL string + LidarrAPIKey string + WebhookSecret string + MusicPath string + DBPath string + Timezone string + WorkWindowEnabled bool + WorkWindowStart string + WorkWindowEnd string + NtfyEnabled bool + NtfyURL string + NtfyTopic string + NtfyToken string + JobTimeoutMinutes int + ExpirerIntervalMinutes int } func Load() (*Config, error) { @@ -83,25 +85,27 @@ func Build(overrides map[string]string) *Config { func buildConfig(overrides map[string]string) *Config { return &Config{ - Port: mergeString(overrides, "PORT", "8080"), - Workers: mergeInt(overrides, "WORKERS", 4), - FFmpegBin: mergeString(overrides, "FFMPEG_BIN", "/usr/bin/ffmpeg"), - FFprobeBin: mergeString(overrides, "FFPROBE_BIN", "/usr/bin/ffprobe"), - LogLevel: mergeString(overrides, "LOG_LEVEL", "INFO"), - DryRun: mergeBool(overrides, "DRY_RUN", false), - LidarrURL: mergeString(overrides, "LIDARR_URL", ""), - LidarrAPIKey: mergeString(overrides, "LIDARR_API_KEY", ""), - WebhookSecret: mergeString(overrides, "WEBHOOK_SECRET", ""), - MusicPath: mergeString(overrides, "MUSIC_PATH", "/music"), - DBPath: mergeString(overrides, "DB_PATH", "/data/coda.db"), - Timezone: mergeString(overrides, "TIMEZONE", "UTC"), - WorkWindowEnabled: mergeBool(overrides, "WORK_WINDOW_ENABLED", false), - WorkWindowStart: mergeString(overrides, "WORK_WINDOW_START", "08:00"), - WorkWindowEnd: mergeString(overrides, "WORK_WINDOW_END", "18:00"), - NtfyEnabled: mergeBool(overrides, "NTFY_ENABLED", false), - NtfyURL: mergeString(overrides, "NTFY_URL", ""), - NtfyTopic: mergeString(overrides, "NTFY_TOPIC", ""), - NtfyToken: mergeString(overrides, "NTFY_TOKEN", ""), + Port: mergeString(overrides, "PORT", "8080"), + Workers: mergeInt(overrides, "WORKERS", 4), + FFmpegBin: mergeString(overrides, "FFMPEG_BIN", "/usr/bin/ffmpeg"), + FFprobeBin: mergeString(overrides, "FFPROBE_BIN", "/usr/bin/ffprobe"), + LogLevel: mergeString(overrides, "LOG_LEVEL", "INFO"), + DryRun: mergeBool(overrides, "DRY_RUN", false), + LidarrURL: mergeString(overrides, "LIDARR_URL", ""), + LidarrAPIKey: mergeString(overrides, "LIDARR_API_KEY", ""), + WebhookSecret: mergeString(overrides, "WEBHOOK_SECRET", ""), + MusicPath: mergeString(overrides, "MUSIC_PATH", "/music"), + DBPath: mergeString(overrides, "DB_PATH", "/data/coda.db"), + Timezone: mergeString(overrides, "TIMEZONE", "UTC"), + WorkWindowEnabled: mergeBool(overrides, "WORK_WINDOW_ENABLED", false), + WorkWindowStart: mergeString(overrides, "WORK_WINDOW_START", "08:00"), + WorkWindowEnd: mergeString(overrides, "WORK_WINDOW_END", "18:00"), + NtfyEnabled: mergeBool(overrides, "NTFY_ENABLED", false), + NtfyURL: mergeString(overrides, "NTFY_URL", ""), + NtfyTopic: mergeString(overrides, "NTFY_TOPIC", ""), + NtfyToken: mergeString(overrides, "NTFY_TOKEN", ""), + JobTimeoutMinutes: mergeInt(overrides, "JOB_TIMEOUT_MINUTES", 120), + ExpirerIntervalMinutes: mergeInt(overrides, "EXPIRER_INTERVAL_MINUTES", 10), } } diff --git a/internal/expirer/expirer.go b/internal/expirer/expirer.go new file mode 100644 index 0000000..c2c9fe3 --- /dev/null +++ b/internal/expirer/expirer.go @@ -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) + } +} diff --git a/internal/models/models.go b/internal/models/models.go index 493ae3a..234209b 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -9,6 +9,7 @@ const ( StatusProcessing JobStatus = "processing" StatusDone JobStatus = "done" StatusError JobStatus = "error" + StatusTimeout JobStatus = "timeout" ) type JobSource string diff --git a/internal/store/jobs.go b/internal/store/jobs.go index 5e15967..7d31c85 100644 --- a/internal/store/jobs.go +++ b/internal/store/jobs.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "fmt" + "strings" "time" "coda/internal/models" @@ -145,3 +146,35 @@ func (r *JobRepository) Finish(ctx context.Context, id string, status models.Job } 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 +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index 03fe78d..543df7d 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -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 { diff --git a/internal/web/dashboard.go b/internal/web/dashboard.go index d1dcab0..ade2a0b 100644 --- a/internal/web/dashboard.go +++ b/internal/web/dashboard.go @@ -23,6 +23,7 @@ type dashboardData struct { ProcessingCount int DoneCount int ErrorCount int + TimeoutCount int } 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.DoneCount = h.jobCount(ctx, models.StatusDone) dd.ErrorCount = h.jobCount(ctx, models.StatusError) + dd.TimeoutCount = h.jobCount(ctx, models.StatusTimeout) 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), DoneCount: h.jobCount(ctx, models.StatusDone), ErrorCount: h.jobCount(ctx, models.StatusError), + TimeoutCount: h.jobCount(ctx, models.StatusTimeout), } h.render.Render(w, "status.html", dd) diff --git a/internal/web/render.go b/internal/web/render.go index 0cd86c5..0539466 100644 --- a/internal/web/render.go +++ b/internal/web/render.go @@ -81,6 +81,8 @@ func statusBadge(level string) template.HTML { return template.HTML(`` + level + ``) case "received": return template.HTML(`` + level + ``) + case "timeout": + return template.HTML(`` + level + ``) default: return template.HTML(level) } diff --git a/web/templates/dashboard.html b/web/templates/dashboard.html index 00cb2cf..6ac1a0b 100644 --- a/web/templates/dashboard.html +++ b/web/templates/dashboard.html @@ -31,7 +31,7 @@ -
+

Received

@@ -56,6 +56,12 @@

{{ .ErrorCount }}

+
+
+

Timeout

+

{{ .TimeoutCount }}

+
+