diff --git a/cmd/agent/cancel_test.go b/cmd/agent/cancel_test.go
new file mode 100644
index 0000000..63e7eaf
--- /dev/null
+++ b/cmd/agent/cancel_test.go
@@ -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)
+ }
+}
diff --git a/cmd/agent/main.go b/cmd/agent/main.go
index 96aedff..f500746 100644
--- a/cmd/agent/main.go
+++ b/cmd/agent/main.go
@@ -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
diff --git a/internal/server/http/job_abandon.go b/internal/server/http/job_abandon.go
new file mode 100644
index 0000000..36d5bf5
--- /dev/null
+++ b/internal/server/http/job_abandon.go
@@ -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)
+}
diff --git a/internal/server/http/job_abandon_test.go b/internal/server/http/job_abandon_test.go
new file mode 100644
index 0000000..1b156b7
--- /dev/null
+++ b/internal/server/http/job_abandon_test.go
@@ -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)
+ }
+}
diff --git a/internal/server/http/server.go b/internal/server/http/server.go
index af63427..24641e8 100644
--- a/internal/server/http/server.go
+++ b/internal/server/http/server.go
@@ -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)
}
})
diff --git a/internal/server/ws/handler.go b/internal/server/ws/handler.go
index 88b9ec9..d2a4a87 100644
--- a/internal/server/ws/handler.go
+++ b/internal/server/ws/handler.go
@@ -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
diff --git a/internal/server/ws/handler_test.go b/internal/server/ws/handler_test.go
index 1bd2088..7f68da8 100644
--- a/internal/server/ws/handler_test.go
+++ b/internal/server/ws/handler_test.go
@@ -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)
diff --git a/web/templates/pages/alerts.html b/web/templates/pages/alerts.html
index 10a8dbc..aa37a5c 100644
--- a/web/templates/pages/alerts.html
+++ b/web/templates/pages/alerts.html
@@ -24,6 +24,10 @@
diff --git a/web/templates/partials/alert_row.html b/web/templates/partials/alert_row.html
index 0ac05db..dc12272 100644
--- a/web/templates/partials/alert_row.html
+++ b/web/templates/partials/alert_row.html
@@ -66,6 +66,14 @@
{{/* actions */}}
{{if eq $status "open"}}
+ {{if eq $a.Kind "job_stuck"}}
+
+ {{end}}