Files
restic-manager/internal/store/pending.go
T
steve e871b05b38
CI / Test (linux/amd64) (pull_request) Successful in 34s
CI / Lint (pull_request) Failing after 16s
CI / Build (windows/amd64) (pull_request) Successful in 22s
CI / Build (linux/amd64) (pull_request) Successful in 20s
CI / Build (linux/arm64) (pull_request) Successful in 21s
lint: drive baseline to zero, drop only-new-issues gate
Cleanup pass over the repo so CI can enforce lint going forward
without the only-new-issues escape hatch:

* gofumpt -w across the tree (31 hits, all formatting)
* misspell --fix (25 hits, US-locale spelling) — but reverted on
  api.JobCancelled = "cancelled" since that literal is the wire +
  DB CHECK constraint value, plus matched the case in store/fleet.go
  back to "cancelled" and added //nolint:misspell on both for the
  next time someone reaches for the auto-fix
* Wrap every `defer rows.Close()` / `defer stmt.Close()` /
  `defer res.Body.Close()` in `defer func() { _ = .Close() }()`
  to satisfy errcheck without losing the close itself
* websocket.Dial callers (1 prod, 4 tests) now capture + close the
  upgrade response Body — coder/websocket can return res with a nil
  Body on success, so the test deferred-closes guard against that
* Annotate the two genuine-by-design nilerr cases with //nolint
  comments explaining why nil-on-error is the contract (cookie
  missing = no session; ctx cancelled mid-backoff = clean shutdown)
* Add brief godoc on the 10 exported const groups + types that
  revive flagged (api.HostOS/HostArch/JobKind/JobStatus/LogStream/
  ErrorCode, restic.EventKind, store.Role, web.FS)
* Drop the unused (*Server).userByID method
* Inline the unparam baseView(active) — every UI page is under
  the dashboard primary nav today

Result: `golangci-lint run ./...` reports 0 issues. CI lint job
no longer needs only-new-issues: true; X-06 follow-up entry in
tasks.md removed.
2026-05-03 16:15:17 +01:00

104 lines
3.2 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package store
import (
"context"
"errors"
"fmt"
"time"
)
// EnqueuePendingRun queues a missed cron tick for the offline-retry
// ticker to dispatch later. Caller (the schedule firing path) sets
// next_attempt_at = now + group.retry_backoff_seconds × 2^(attempt-1).
func (st *Store) EnqueuePendingRun(ctx context.Context, p *PendingRun) error {
if p.ID == "" || p.ScheduleID == "" || p.SourceGroupID == "" || p.HostID == "" {
return errors.New("store: pending run id, schedule_id, source_group_id, host_id required")
}
if p.Attempt == 0 {
p.Attempt = 1
}
if p.NextAttemptAt.IsZero() {
p.NextAttemptAt = time.Now().UTC()
}
if p.ScheduledAt.IsZero() {
p.ScheduledAt = time.Now().UTC()
}
_, err := st.db.ExecContext(ctx,
`INSERT INTO pending_runs (id, schedule_id, source_group_id, host_id,
attempt, next_attempt_at, scheduled_at, last_error)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
p.ID, p.ScheduleID, p.SourceGroupID, p.HostID,
p.Attempt,
p.NextAttemptAt.UTC().Format(time.RFC3339Nano),
p.ScheduledAt.UTC().Format(time.RFC3339Nano),
nullableString(p.LastError))
if err != nil {
return fmt.Errorf("store: enqueue pending run: %w", err)
}
return nil
}
// DuePendingRuns returns rows whose next_attempt_at <= now, ordered
// oldest first. Server-side ticker calls this every ~30s.
func (st *Store) DuePendingRuns(ctx context.Context, now time.Time, limit int) ([]PendingRun, error) {
rows, err := st.db.QueryContext(ctx,
`SELECT id, schedule_id, source_group_id, host_id, attempt,
next_attempt_at, scheduled_at, COALESCE(last_error, '')
FROM pending_runs
WHERE next_attempt_at <= ?
ORDER BY next_attempt_at
LIMIT ?`,
now.UTC().Format(time.RFC3339Nano), limit)
if err != nil {
return nil, fmt.Errorf("store: due pending runs: %w", err)
}
defer func() { _ = rows.Close() }()
out := []PendingRun{}
for rows.Next() {
var p PendingRun
var nextAt, scheduledAt string
if err := rows.Scan(&p.ID, &p.ScheduleID, &p.SourceGroupID, &p.HostID,
&p.Attempt, &nextAt, &scheduledAt, &p.LastError); err != nil {
return nil, err
}
if t, err := time.Parse(time.RFC3339Nano, nextAt); err == nil {
p.NextAttemptAt = t
}
if t, err := time.Parse(time.RFC3339Nano, scheduledAt); err == nil {
p.ScheduledAt = t
}
out = append(out, p)
}
return out, rows.Err()
}
// DeletePendingRun removes a row by id. Called after successful
// dispatch or after exceeding retry_max.
func (st *Store) DeletePendingRun(ctx context.Context, id string) error {
_, err := st.db.ExecContext(ctx,
`DELETE FROM pending_runs WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("store: delete pending run: %w", err)
}
return nil
}
// BumpPendingRunAttempt increments the attempt counter and updates
// next_attempt_at + last_error. Used after a failed retry — caller
// has decided to try again.
func (st *Store) BumpPendingRunAttempt(ctx context.Context, id string, nextAttemptAt time.Time, lastError string) error {
_, err := st.db.ExecContext(ctx,
`UPDATE pending_runs SET
attempt = attempt + 1,
next_attempt_at = ?,
last_error = ?
WHERE id = ?`,
nextAttemptAt.UTC().Format(time.RFC3339Nano),
nullableString(lastError),
id)
if err != nil {
return fmt.Errorf("store: bump pending run: %w", err)
}
return nil
}