Compare commits
4 Commits
v1.2.1
...
5fb08ba489
| Author | SHA1 | Date | |
|---|---|---|---|
| 5fb08ba489 | |||
| 4f56b8f705 | |||
| 25866ff102 | |||
| ec448157d9 |
@@ -0,0 +1,68 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
|
||||
)
|
||||
|
||||
type recordingSender struct {
|
||||
envelopes []api.Envelope
|
||||
}
|
||||
|
||||
func (s *recordingSender) Send(env api.Envelope) error {
|
||||
s.envelopes = append(s.envelopes, env)
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestCancelCommandReportsUnknownJob(t *testing.T) {
|
||||
d := &dispatcher{}
|
||||
tx := &recordingSender{}
|
||||
env, err := api.Marshal(api.MsgCommandCancel, "request-1", api.CommandCancelPayload{JobID: "missing-job"})
|
||||
if err != nil {
|
||||
t.Fatalf("marshal command: %v", err)
|
||||
}
|
||||
|
||||
if err := d.handle(context.Background(), env, tx); err != nil {
|
||||
t.Fatalf("handle cancel: %v", err)
|
||||
}
|
||||
if len(tx.envelopes) != 1 {
|
||||
t.Fatalf("sent %d envelopes, want 1", len(tx.envelopes))
|
||||
}
|
||||
if tx.envelopes[0].Type != api.MsgCommandResult || tx.envelopes[0].ID != env.ID {
|
||||
t.Fatalf("unexpected result envelope: %+v", tx.envelopes[0])
|
||||
}
|
||||
var result api.CommandResultPayload
|
||||
if err := tx.envelopes[0].UnmarshalPayload(&result); err != nil {
|
||||
t.Fatalf("unmarshal result: %v", err)
|
||||
}
|
||||
if result.JobID != "missing-job" || result.Accepted || result.Error != "job_not_found" {
|
||||
t.Fatalf("unexpected result: %+v", result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCancelCommandCancelsTrackedJob(t *testing.T) {
|
||||
d := &dispatcher{}
|
||||
cancelled := false
|
||||
d.trackJob("running-job", func() { cancelled = true })
|
||||
tx := &recordingSender{}
|
||||
env, err := api.Marshal(api.MsgCommandCancel, "request-2", api.CommandCancelPayload{JobID: "running-job"})
|
||||
if err != nil {
|
||||
t.Fatalf("marshal command: %v", err)
|
||||
}
|
||||
|
||||
if err := d.handle(context.Background(), env, tx); err != nil {
|
||||
t.Fatalf("handle cancel: %v", err)
|
||||
}
|
||||
if !cancelled {
|
||||
t.Fatal("tracked job was not cancelled")
|
||||
}
|
||||
var result api.CommandResultPayload
|
||||
if err := tx.envelopes[0].UnmarshalPayload(&result); err != nil {
|
||||
t.Fatalf("unmarshal result: %v", err)
|
||||
}
|
||||
if !result.Accepted || result.Error != "" {
|
||||
t.Fatalf("unexpected result: %+v", result)
|
||||
}
|
||||
}
|
||||
+13
-1
@@ -276,7 +276,8 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S
|
||||
if err := env.UnmarshalPayload(&p); err != nil {
|
||||
return fmt.Errorf("command.cancel: %w", err)
|
||||
}
|
||||
if d.cancelJob(p.JobID) {
|
||||
known := d.cancelJob(p.JobID)
|
||||
if known {
|
||||
slog.Info("ws agent: command.cancel applied", "job_id", p.JobID)
|
||||
} else {
|
||||
// Job already finished or was never seen on this agent.
|
||||
@@ -284,6 +285,17 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S
|
||||
// natural completion. Server-side state is authoritative.
|
||||
slog.Info("ws agent: command.cancel for unknown job (already finished?)", "job_id", p.JobID)
|
||||
}
|
||||
result := api.CommandResultPayload{JobID: p.JobID, Accepted: known}
|
||||
if !known {
|
||||
result.Error = "job_not_found"
|
||||
}
|
||||
ack, err := api.Marshal(api.MsgCommandResult, env.ID, result)
|
||||
if err != nil {
|
||||
return fmt.Errorf("command.cancel result: %w", err)
|
||||
}
|
||||
if err := tx.Send(ack); err != nil {
|
||||
return fmt.Errorf("command.cancel result send: %w", err)
|
||||
}
|
||||
|
||||
case api.MsgTreeList:
|
||||
// Synchronous RPC for the restore wizard's tree browser. The
|
||||
|
||||
@@ -6,11 +6,12 @@ Three ways to trigger one:
|
||||
|
||||
1. **Scheduled** — the agent's local cron fires at the time set
|
||||
on the schedule.
|
||||
2. **Run-now** — operator clicks **Run now** on the host detail
|
||||
right rail. Posts to `/hosts/{id}/run-backup` (defaults to all
|
||||
source groups) or to a per-group form for finer control.
|
||||
3. **API** — `POST /api/hosts/{id}/jobs` with the appropriate
|
||||
payload. Same audit + dispatch path.
|
||||
2. **Run-now** — operator clicks **Run now** for a specific source
|
||||
group. This uses `POST /hosts/{id}/source-groups/{gid}/run`.
|
||||
3. **API** — use `POST /api/hosts/{id}/source-groups/{gid}/run` for a
|
||||
configured source group. The lower-level `POST /api/hosts/{id}/jobs`
|
||||
backup form requires explicit paths in `args`; an empty backup is
|
||||
rejected instead of dispatching a job that restic cannot run.
|
||||
|
||||
In every case the server creates a `jobs` row, broadcasts a
|
||||
`command.run` to the host, and lands the operator on the live
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
stdhttp "net/http"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/oklog/ulid/v2"
|
||||
|
||||
"gitea.dcglab.co.uk/steve/restic-manager/internal/alert"
|
||||
"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
|
||||
"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
|
||||
)
|
||||
|
||||
func (s *Server) abandonJob(ctx context.Context, userID *string, jobID, reason string) error {
|
||||
job, err := s.deps.Store.GetJob(ctx, jobID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
switch api.JobStatus(job.Status) {
|
||||
case api.JobSucceeded, api.JobFailed, api.JobCancelled:
|
||||
return errJobTerminal
|
||||
}
|
||||
when := time.Now().UTC()
|
||||
if err := s.deps.Store.MarkJobFinished(ctx, jobID, string(api.JobCancelled), -1, nil, reason, when); err != nil {
|
||||
return err
|
||||
}
|
||||
_ = s.deps.Store.AppendAudit(ctx, store.AuditEntry{
|
||||
ID: ulid.Make().String(), UserID: userID, Actor: "user",
|
||||
Action: "job.abandon", TargetKind: ptr("job"), TargetID: &jobID,
|
||||
TS: when,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
var errJobTerminal = errors.New("job already terminal")
|
||||
|
||||
func (s *Server) handleAbandonJob(w stdhttp.ResponseWriter, r *stdhttp.Request) {
|
||||
user, ok := s.requireUser(r)
|
||||
if !ok {
|
||||
writeJSONError(w, stdhttp.StatusUnauthorized, "unauthorised", "")
|
||||
return
|
||||
}
|
||||
jobID := chi.URLParam(r, "id")
|
||||
err := s.abandonJob(r.Context(), &user.ID, jobID, "abandoned by operator")
|
||||
switch {
|
||||
case errors.Is(err, store.ErrNotFound):
|
||||
writeJSONError(w, stdhttp.StatusNotFound, "job_not_found", "")
|
||||
case errors.Is(err, errJobTerminal):
|
||||
writeJSONError(w, stdhttp.StatusConflict, "job_terminal", "job is already terminal")
|
||||
case err != nil:
|
||||
writeJSONError(w, stdhttp.StatusInternalServerError, "internal", "")
|
||||
default:
|
||||
writeJSON(w, stdhttp.StatusOK, map[string]string{"job_id": jobID, "status": string(api.JobCancelled)})
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) abandonStuckAlerts(ctx context.Context, userID *string) (int, error) {
|
||||
open, err := s.deps.Store.ListAlerts(ctx, store.AlertFilter{Status: "open"})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
acked, err := s.deps.Store.ListAlerts(ctx, store.AlertFilter{Status: "acknowledged"})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
count := 0
|
||||
seen := map[string]bool{}
|
||||
for _, a := range append(open, acked...) {
|
||||
if a.Kind != alert.KindJobStuck || seen[a.DedupKey] {
|
||||
continue
|
||||
}
|
||||
seen[a.DedupKey] = true
|
||||
err := s.abandonJob(ctx, userID, a.DedupKey, "abandoned by operator after stuck-job alert")
|
||||
if err != nil {
|
||||
if errors.Is(err, errJobTerminal) || errors.Is(err, store.ErrNotFound) {
|
||||
s.resolveAlert(ctx, a.ID)
|
||||
continue
|
||||
}
|
||||
return count, err
|
||||
}
|
||||
count++
|
||||
s.resolveAlert(ctx, a.ID)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
func (s *Server) resolveAlert(ctx context.Context, alertID string) {
|
||||
when := time.Now().UTC()
|
||||
if s.deps.AlertEngine != nil {
|
||||
_ = s.deps.AlertEngine.Resolve(ctx, alertID, when)
|
||||
return
|
||||
}
|
||||
_ = s.deps.Store.Resolve(ctx, alertID, when)
|
||||
}
|
||||
|
||||
func (s *Server) handleAbandonStuckJobs(w stdhttp.ResponseWriter, r *stdhttp.Request) {
|
||||
user, ok := s.requireUser(r)
|
||||
if !ok {
|
||||
writeJSONError(w, stdhttp.StatusUnauthorized, "unauthorised", "")
|
||||
return
|
||||
}
|
||||
count, err := s.abandonStuckAlerts(r.Context(), &user.ID)
|
||||
if err != nil {
|
||||
writeJSONError(w, stdhttp.StatusInternalServerError, "internal", "")
|
||||
return
|
||||
}
|
||||
writeJSON(w, stdhttp.StatusOK, map[string]int{"abandoned": count})
|
||||
}
|
||||
|
||||
func (s *Server) handleUIAbandonAlertJob(w stdhttp.ResponseWriter, r *stdhttp.Request) {
|
||||
user := s.requireUIUser(w, r)
|
||||
if user == nil {
|
||||
return
|
||||
}
|
||||
alertID := chi.URLParam(r, "id")
|
||||
a, err := s.deps.Store.GetAlert(r.Context(), alertID)
|
||||
if err != nil || a == nil || a.Kind != alert.KindJobStuck {
|
||||
stdhttp.Error(w, "stuck-job alert not found", stdhttp.StatusNotFound)
|
||||
return
|
||||
}
|
||||
if err := s.abandonJob(r.Context(), &user.ID, a.DedupKey, "abandoned by operator after stuck-job alert"); err != nil && !errors.Is(err, errJobTerminal) {
|
||||
stdhttp.Error(w, "unable to abandon job", stdhttp.StatusConflict)
|
||||
return
|
||||
}
|
||||
s.resolveAlert(r.Context(), alertID)
|
||||
w.Header().Set("HX-Redirect", "/alerts?"+r.URL.RawQuery)
|
||||
w.WriteHeader(stdhttp.StatusNoContent)
|
||||
}
|
||||
|
||||
func (s *Server) handleUIAbandonStuckJobs(w stdhttp.ResponseWriter, r *stdhttp.Request) {
|
||||
user := s.requireUIUser(w, r)
|
||||
if user == nil {
|
||||
return
|
||||
}
|
||||
if _, err := s.abandonStuckAlerts(r.Context(), &user.ID); err != nil {
|
||||
stdhttp.Error(w, "unable to abandon stuck jobs", stdhttp.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
stdhttp.Redirect(w, r, "/alerts", stdhttp.StatusSeeOther)
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
package http
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
stdhttp "net/http"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/oklog/ulid/v2"
|
||||
|
||||
"gitea.dcglab.co.uk/steve/restic-manager/internal/alert"
|
||||
"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
|
||||
)
|
||||
|
||||
func seedRunningJob(t *testing.T, st *store.Store, hostID string) string {
|
||||
t.Helper()
|
||||
id := ulid.Make().String()
|
||||
when := time.Now().UTC().Add(-24 * time.Hour)
|
||||
if err := st.CreateJob(context.Background(), store.Job{
|
||||
ID: id, HostID: hostID, Kind: "backup", ActorKind: "user", CreatedAt: when,
|
||||
}); err != nil {
|
||||
t.Fatalf("create job: %v", err)
|
||||
}
|
||||
if err := st.MarkJobStarted(context.Background(), id, when); err != nil {
|
||||
t.Fatalf("start job: %v", err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func TestAbandonJobMakesOrphanTerminalAndAudits(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, ts, st := rawTestServer(t)
|
||||
hostID := makeHost(t, st, "abandon-host")
|
||||
jobID := seedRunningJob(t, st, hostID)
|
||||
req, _ := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/jobs/"+jobID+"/abandon", nil)
|
||||
req.AddCookie(loginAsAdmin(t, st))
|
||||
res, err := stdhttp.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("abandon: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
if res.StatusCode != stdhttp.StatusOK {
|
||||
t.Fatalf("status: got %d", res.StatusCode)
|
||||
}
|
||||
job, _ := st.GetJob(context.Background(), jobID)
|
||||
if job.Status != "cancelled" || job.FinishedAt == nil || job.Error == nil {
|
||||
t.Fatalf("job not terminalised: %+v", job)
|
||||
}
|
||||
var audits int
|
||||
_ = st.DB().QueryRow(`SELECT COUNT(*) FROM audit_log WHERE action = 'job.abandon' AND target_id = ?`, jobID).Scan(&audits)
|
||||
if audits != 1 {
|
||||
t.Fatalf("audit rows: got %d", audits)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBulkAbandonTargetsOnlyAlertedStuckJobs(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, ts, st := rawTestServer(t)
|
||||
hostID := makeHost(t, st, "bulk-abandon-host")
|
||||
stuckID := seedRunningJob(t, st, hostID)
|
||||
unalertedID := seedRunningJob(t, st, hostID)
|
||||
_, _, err := st.RaiseOrTouch(context.Background(), hostID, alert.KindJobStuck, stuckID,
|
||||
"warning", "stuck", time.Now().UTC())
|
||||
if err != nil {
|
||||
t.Fatalf("raise alert: %v", err)
|
||||
}
|
||||
req, _ := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/jobs/abandon-stuck", bytes.NewReader(nil))
|
||||
req.AddCookie(loginAsAdmin(t, st))
|
||||
res, err := stdhttp.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("bulk abandon: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
if res.StatusCode != stdhttp.StatusOK {
|
||||
t.Fatalf("status: got %d", res.StatusCode)
|
||||
}
|
||||
var out map[string]int
|
||||
_ = json.NewDecoder(res.Body).Decode(&out)
|
||||
if out["abandoned"] != 1 {
|
||||
t.Fatalf("abandoned: %+v", out)
|
||||
}
|
||||
stuck, _ := st.GetJob(context.Background(), stuckID)
|
||||
unalerted, _ := st.GetJob(context.Background(), unalertedID)
|
||||
if stuck.Status != "cancelled" || unalerted.Status != "running" {
|
||||
t.Fatalf("statuses: stuck=%s unalerted=%s", stuck.Status, unalerted.Status)
|
||||
}
|
||||
open, _ := st.ListAlerts(context.Background(), store.AlertFilter{Status: "open"})
|
||||
if len(open) != 0 {
|
||||
t.Fatalf("stuck alerts remained open: %+v", open)
|
||||
}
|
||||
}
|
||||
@@ -69,6 +69,10 @@ func (s *Server) dispatchJob(ctx context.Context, user *store.User,
|
||||
Kind: kind,
|
||||
Args: args,
|
||||
}
|
||||
if kind == api.JobBackup && len(args) == 0 {
|
||||
return res, stdhttp.StatusUnprocessableEntity, "backup_paths_required",
|
||||
"backup requires paths in args; configured backups must use POST /api/hosts/{id}/source-groups/{gid}/run"
|
||||
}
|
||||
if kind == api.JobForget {
|
||||
if !validForgetArgs(args) {
|
||||
return res, stdhttp.StatusBadRequest, "invalid_args",
|
||||
|
||||
@@ -14,6 +14,34 @@ import (
|
||||
"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
|
||||
)
|
||||
|
||||
func TestRunNowBackupRejectsEmptyPaths(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, ts, st := rawTestServer(t)
|
||||
cookie := loginAsAdmin(t, st)
|
||||
hostID := makeHost(t, st, "empty-backup-host")
|
||||
req, _ := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/hosts/"+hostID+"/jobs",
|
||||
bytes.NewReader([]byte(`{"kind":"backup"}`)))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.AddCookie(cookie)
|
||||
res, err := stdhttp.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("post run-now: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
if res.StatusCode != stdhttp.StatusUnprocessableEntity {
|
||||
t.Fatalf("status: got %d, want 422", res.StatusCode)
|
||||
}
|
||||
body := readJSONError(t, res.Body)
|
||||
if body.Code != "backup_paths_required" {
|
||||
t.Fatalf("code: got %q", body.Code)
|
||||
}
|
||||
var jobs int
|
||||
_ = st.DB().QueryRow(`SELECT COUNT(*) FROM jobs WHERE host_id = ?`, hostID).Scan(&jobs)
|
||||
if jobs != 0 {
|
||||
t.Fatalf("invalid backup created %d jobs", jobs)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunNowForgetShipsRetentionGroupsAndDryRun(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv, ts, st := rawTestServer(t)
|
||||
|
||||
@@ -153,6 +153,8 @@ func TestSourceGroupsCRUD(t *testing.T) {
|
||||
},
|
||||
"retry_max": 3,
|
||||
"retry_backoff_seconds": 60,
|
||||
"pre_hook": "prepare-db",
|
||||
"post_hook": "resume-db",
|
||||
}, cookie)
|
||||
if status != 201 {
|
||||
t.Fatalf("create status: %d, body: %+v", status, body)
|
||||
@@ -161,6 +163,16 @@ func TestSourceGroupsCRUD(t *testing.T) {
|
||||
if gid == "" {
|
||||
t.Fatalf("create: no id returned: %+v", body)
|
||||
}
|
||||
if body["has_pre_hook"] != true || body["has_post_hook"] != true {
|
||||
t.Fatalf("create hook indicators: %+v", body)
|
||||
}
|
||||
stored, err := st.GetSourceGroup(context.Background(), hostID, gid)
|
||||
if err != nil {
|
||||
t.Fatalf("get stored group: %v", err)
|
||||
}
|
||||
if stored.PreHook == "prepare-db" || stored.PostHook == "resume-db" || stored.PreHook == "" || stored.PostHook == "" {
|
||||
t.Fatalf("hooks not encrypted at rest: pre=%q post=%q", stored.PreHook, stored.PostHook)
|
||||
}
|
||||
|
||||
// Duplicate name → 409.
|
||||
status, _ = doJSON(t, url, "POST", "/api/hosts/"+hostID+"/source-groups",
|
||||
@@ -185,6 +197,20 @@ func TestSourceGroupsCRUD(t *testing.T) {
|
||||
if got := body["name"]; got != "system" {
|
||||
t.Errorf("rename: got %v want system", got)
|
||||
}
|
||||
stored, _ = st.GetSourceGroup(context.Background(), hostID, gid)
|
||||
if stored.PreHook == "" || stored.PostHook == "" {
|
||||
t.Fatal("PUT without hook fields cleared existing hooks")
|
||||
}
|
||||
|
||||
// Explicit empty hook clears only that hook; plaintext is never returned.
|
||||
status, body = doJSON(t, url, "PUT", "/api/hosts/"+hostID+"/source-groups/"+gid,
|
||||
map[string]any{"name": "system", "includes": []string{"/etc"}, "pre_hook": ""}, cookie)
|
||||
if status != 200 || body["has_pre_hook"] != false || body["has_post_hook"] != true {
|
||||
t.Fatalf("clear hook status=%d body=%+v", status, body)
|
||||
}
|
||||
if _, exposed := body["post_hook"]; exposed {
|
||||
t.Fatal("source-group response exposed hook plaintext field")
|
||||
}
|
||||
|
||||
// Delete.
|
||||
status, _ = doJSON(t, url, "DELETE", "/api/hosts/"+hostID+"/source-groups/"+gid, nil, cookie)
|
||||
|
||||
@@ -261,6 +261,8 @@ func (s *Server) routes(r chi.Router) {
|
||||
r.Post("/api/hosts/{id}/repo/check", s.handleRunRepoCheck)
|
||||
r.Post("/api/hosts/{id}/repo/unlock", s.handleRunRepoUnlock)
|
||||
r.Post("/api/jobs/{id}/cancel", s.handleCancelJob)
|
||||
r.Post("/api/jobs/{id}/abandon", s.handleAbandonJob)
|
||||
r.Post("/api/jobs/abandon-stuck", s.handleAbandonStuckJobs)
|
||||
r.Post("/api/hosts/{id}/snapshots/diff", s.handleSnapshotDiff)
|
||||
r.Post("/api/hosts/{id}/snapshots/refresh", s.handleRefreshHostSnapshots)
|
||||
|
||||
@@ -298,6 +300,8 @@ func (s *Server) routes(r chi.Router) {
|
||||
r.Post("/hosts/{id}/restore", s.handleUIRestorePost)
|
||||
r.Post("/alerts/{id}/acknowledge", s.handleUIAlertAcknowledge)
|
||||
r.Post("/alerts/{id}/resolve", s.handleUIAlertResolve)
|
||||
r.Post("/alerts/{id}/abandon-job", s.handleUIAbandonAlertJob)
|
||||
r.Post("/alerts/abandon-stuck", s.handleUIAbandonStuckJobs)
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
@@ -29,6 +29,8 @@ type sourceGroupView struct {
|
||||
RetryMax int `json:"retry_max"`
|
||||
RetryBackoffSeconds int `json:"retry_backoff_seconds"`
|
||||
ConflictDimension string `json:"conflict_dimension,omitempty"`
|
||||
HasPreHook bool `json:"has_pre_hook"`
|
||||
HasPostHook bool `json:"has_post_hook"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
@@ -49,6 +51,8 @@ func toSourceGroupView(g store.SourceGroup) sourceGroupView {
|
||||
RetryMax: g.RetryMax,
|
||||
RetryBackoffSeconds: g.RetryBackoffSeconds,
|
||||
ConflictDimension: g.ConflictDimension,
|
||||
HasPreHook: g.PreHook != "",
|
||||
HasPostHook: g.PostHook != "",
|
||||
CreatedAt: g.CreatedAt,
|
||||
UpdatedAt: g.UpdatedAt,
|
||||
}
|
||||
@@ -62,6 +66,29 @@ type sourceGroupWriteRequest struct {
|
||||
RetentionPolicy store.RetentionPolicy `json:"retention_policy"`
|
||||
RetryMax int `json:"retry_max"`
|
||||
RetryBackoffSeconds int `json:"retry_backoff_seconds"`
|
||||
// Pointer fields distinguish omission (preserve on PUT) from an explicit
|
||||
// empty string (clear). Hook plaintext is accepted on write but never
|
||||
// returned by the JSON API.
|
||||
PreHook *string `json:"pre_hook,omitempty"`
|
||||
PostHook *string `json:"post_hook,omitempty"`
|
||||
}
|
||||
|
||||
func (s *Server) applySourceGroupHooks(hostID string, req sourceGroupWriteRequest, g *store.SourceGroup) error {
|
||||
if req.PreHook != nil {
|
||||
enc, err := s.EncryptHookForGroup(hostID, "pre", *req.PreHook)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
g.PreHook = enc
|
||||
}
|
||||
if req.PostHook != nil {
|
||||
enc, err := s.EncryptHookForGroup(hostID, "post", *req.PostHook)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
g.PostHook = enc
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) handleListSourceGroups(w stdhttp.ResponseWriter, r *stdhttp.Request) {
|
||||
@@ -142,6 +169,10 @@ func (s *Server) handleCreateSourceGroup(w stdhttp.ResponseWriter, r *stdhttp.Re
|
||||
RetryMax: req.RetryMax,
|
||||
RetryBackoffSeconds: req.RetryBackoffSeconds,
|
||||
}
|
||||
if err := s.applySourceGroupHooks(hostID, req, &g); err != nil {
|
||||
writeJSONError(w, stdhttp.StatusInternalServerError, "hook_encryption_failed", "")
|
||||
return
|
||||
}
|
||||
if err := s.deps.Store.CreateSourceGroup(r.Context(), &g); err != nil {
|
||||
writeJSONError(w, stdhttp.StatusInternalServerError, "internal", err.Error())
|
||||
return
|
||||
@@ -157,7 +188,8 @@ func (s *Server) handleUpdateSourceGroup(w stdhttp.ResponseWriter, r *stdhttp.Re
|
||||
}
|
||||
hostID := chi.URLParam(r, "id")
|
||||
groupID := chi.URLParam(r, "gid")
|
||||
if _, err := s.deps.Store.GetSourceGroup(r.Context(), hostID, groupID); err != nil {
|
||||
existingGroup, err := s.deps.Store.GetSourceGroup(r.Context(), hostID, groupID)
|
||||
if err != nil {
|
||||
if errors.Is(err, store.ErrNotFound) {
|
||||
writeJSONError(w, stdhttp.StatusNotFound, "group_not_found", "")
|
||||
return
|
||||
@@ -188,6 +220,12 @@ func (s *Server) handleUpdateSourceGroup(w stdhttp.ResponseWriter, r *stdhttp.Re
|
||||
RetentionPolicy: req.RetentionPolicy,
|
||||
RetryMax: req.RetryMax,
|
||||
RetryBackoffSeconds: req.RetryBackoffSeconds,
|
||||
PreHook: existingGroup.PreHook,
|
||||
PostHook: existingGroup.PostHook,
|
||||
}
|
||||
if err := s.applySourceGroupHooks(hostID, req, &g); err != nil {
|
||||
writeJSONError(w, stdhttp.StatusInternalServerError, "hook_encryption_failed", "")
|
||||
return
|
||||
}
|
||||
if err := s.deps.Store.UpdateSourceGroup(r.Context(), &g); err != nil {
|
||||
writeJSONError(w, stdhttp.StatusInternalServerError, "internal", err.Error())
|
||||
|
||||
@@ -357,10 +357,26 @@ func dispatchAgentMessage(ctx context.Context, c *Conn, hostID string, env api.E
|
||||
}
|
||||
|
||||
case api.MsgCommandResult:
|
||||
// TODO(P2): persist command.result acks for "did the agent
|
||||
// accept the dispatch?" forensics. Currently the job lifecycle
|
||||
// (job.started → job.finished) is sufficient signal.
|
||||
slog.Debug("ws msg not yet handled", "type", env.Type, "host_id", hostID)
|
||||
var p api.CommandResultPayload
|
||||
if err := env.UnmarshalPayload(&p); err != nil {
|
||||
slog.Warn("ws: decode command result", "host_id", hostID, "err", err)
|
||||
break
|
||||
}
|
||||
// A cancel for a job unknown to the current agent process can never
|
||||
// produce job.finished. Terminalise the orphan server-side.
|
||||
if !p.Accepted && p.Error == "job_not_found" && p.JobID != "" {
|
||||
job, err := deps.Store.GetJob(ctx, p.JobID)
|
||||
if err != nil || job.HostID != hostID {
|
||||
slog.Warn("ws: reject orphan result for foreign or missing job", "job_id", p.JobID, "host_id", hostID)
|
||||
break
|
||||
}
|
||||
if job.Status == string(api.JobQueued) || job.Status == string(api.JobRunning) {
|
||||
if err := deps.Store.MarkJobFinished(ctx, p.JobID, string(api.JobCancelled), -1, nil,
|
||||
"agent reported job not found during cancellation", time.Now().UTC()); err != nil {
|
||||
slog.Warn("ws: terminalise orphaned job", "job_id", p.JobID, "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
case api.MsgTreeListResult:
|
||||
// Reply to a synchronous tree.list RPC. Route to the waiter
|
||||
|
||||
@@ -36,6 +36,54 @@ func seedHostWS(t *testing.T, s *store.Store, hostID string) {
|
||||
func int64ptrWS(v int64) *int64 { return &v }
|
||||
func boolptrWS(v bool) *bool { return &v }
|
||||
|
||||
func TestUnknownCancelResultTerminalisesOrphanedJob(t *testing.T) {
|
||||
t.Parallel()
|
||||
s := openWSTestStore(t)
|
||||
ctx := context.Background()
|
||||
const hostID = "h-orphan-ws"
|
||||
seedHostWS(t, s, hostID)
|
||||
jobID := "j-orphan-ws"
|
||||
started := time.Now().UTC().Add(-time.Hour)
|
||||
if err := s.CreateJob(ctx, store.Job{ID: jobID, HostID: hostID, Kind: "backup", ActorKind: "user", CreatedAt: started}); err != nil {
|
||||
t.Fatalf("create job: %v", err)
|
||||
}
|
||||
if err := s.MarkJobStarted(ctx, jobID, started); err != nil {
|
||||
t.Fatalf("start job: %v", err)
|
||||
}
|
||||
env, _ := api.Marshal(api.MsgCommandResult, jobID, api.CommandResultPayload{
|
||||
JobID: jobID, Accepted: false, Error: "job_not_found",
|
||||
})
|
||||
dispatchAgentMessage(ctx, nil, hostID, env, HandlerDeps{Store: s})
|
||||
job, _ := s.GetJob(ctx, jobID)
|
||||
if job.Status != "cancelled" || job.FinishedAt == nil {
|
||||
t.Fatalf("orphan not terminalised: %+v", job)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnknownCancelResultCannotTerminaliseAnotherHostsJob(t *testing.T) {
|
||||
t.Parallel()
|
||||
s := openWSTestStore(t)
|
||||
ctx := context.Background()
|
||||
seedHostWS(t, s, "reporting-host")
|
||||
seedHostWS(t, s, "job-host")
|
||||
const jobID = "j-foreign-ws"
|
||||
started := time.Now().UTC().Add(-time.Hour)
|
||||
if err := s.CreateJob(ctx, store.Job{ID: jobID, HostID: "job-host", Kind: "backup", ActorKind: "user", CreatedAt: started}); err != nil {
|
||||
t.Fatalf("create job: %v", err)
|
||||
}
|
||||
if err := s.MarkJobStarted(ctx, jobID, started); err != nil {
|
||||
t.Fatalf("start job: %v", err)
|
||||
}
|
||||
env, _ := api.Marshal(api.MsgCommandResult, jobID, api.CommandResultPayload{
|
||||
JobID: jobID, Accepted: false, Error: "job_not_found",
|
||||
})
|
||||
dispatchAgentMessage(ctx, nil, "reporting-host", env, HandlerDeps{Store: s})
|
||||
job, _ := s.GetJob(ctx, jobID)
|
||||
if job.Status != "running" || job.FinishedAt != nil {
|
||||
t.Fatalf("foreign job was modified: %+v", job)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepoStatsReportPersisted(t *testing.T) {
|
||||
t.Parallel()
|
||||
s := openWSTestStore(t)
|
||||
|
||||
@@ -24,6 +24,10 @@
|
||||
</h1>
|
||||
</div>
|
||||
<div class="flex gap-2">
|
||||
<form method="post" action="/alerts/abandon-stuck"
|
||||
onsubmit="return confirm('Mark every currently alerted stuck job as cancelled? Use this only for jobs no longer running on their agents.');">
|
||||
<button type="submit" class="btn btn-danger">Abandon all stuck jobs</button>
|
||||
</form>
|
||||
<a href="/settings/notifications" class="btn">Channel settings →</a>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -66,6 +66,14 @@
|
||||
{{/* actions */}}
|
||||
<div style="display: flex; gap: 6px; justify-content: flex-end; align-items: center;">
|
||||
{{if eq $status "open"}}
|
||||
{{if eq $a.Kind "job_stuck"}}
|
||||
<form method="post" action="/alerts/{{$a.ID}}/abandon-job">
|
||||
<button type="submit" class="btn btn-danger"
|
||||
hx-post="/alerts/{{$a.ID}}/abandon-job{{if $qs}}?{{$qs}}{{end}}"
|
||||
hx-swap="none"
|
||||
hx-confirm="Mark this job cancelled on the server? Use this only when it is no longer running on the agent.">Abandon job</button>
|
||||
</form>
|
||||
{{end}}
|
||||
<form method="post" action="/alerts/{{$a.ID}}/acknowledge">
|
||||
{{if $qs}}<input type="hidden" name="qs" value="{{$qs}}">{{end}}
|
||||
<button type="submit" class="btn"
|
||||
@@ -89,6 +97,14 @@
|
||||
{{end}}
|
||||
ack'd{{if $ackedBy}} by {{$ackedBy}}{{end}} · {{relTime $a.AcknowledgedAt}}
|
||||
</span>
|
||||
{{if eq $a.Kind "job_stuck"}}
|
||||
<form method="post" action="/alerts/{{$a.ID}}/abandon-job">
|
||||
<button type="submit" class="btn btn-danger"
|
||||
hx-post="/alerts/{{$a.ID}}/abandon-job{{if $qs}}?{{$qs}}{{end}}"
|
||||
hx-swap="none"
|
||||
hx-confirm="Mark this job cancelled on the server? Use this only when it is no longer running on the agent.">Abandon job</button>
|
||||
</form>
|
||||
{{end}}
|
||||
<form method="post" action="/alerts/{{$a.ID}}/resolve">
|
||||
{{if $qs}}<input type="hidden" name="qs" value="{{$qs}}">{{end}}
|
||||
<button type="submit" class="btn"
|
||||
|
||||
Reference in New Issue
Block a user