From f9718e60772000fc8a7fcceac0bd4d5af4d55ad9 Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Sat, 22 Aug 2026 11:00:46 +0100 Subject: [PATCH] Make snapshot projections self-diagnosing --- cmd/agent/main.go | 19 +++++ internal/agent/runner/runner.go | 22 +++-- internal/agent/runner/runner_test.go | 21 ++++- internal/api/wire.go | 13 +-- internal/server/http/server.go | 1 + internal/server/http/snapshots.go | 45 +++++++++- internal/server/http/snapshots_test.go | 83 +++++++++++++++++++ internal/store/hosts.go | 16 +++- internal/store/jobs.go | 21 +++++ .../0025_hosts_snapshot_refreshed_at.sql | 1 + internal/store/snapshots.go | 4 +- internal/store/snapshots_test.go | 6 +- internal/store/types.go | 1 + 13 files changed, 224 insertions(+), 29 deletions(-) create mode 100644 internal/server/http/snapshots_test.go create mode 100644 internal/store/migrations/0025_hosts_snapshot_refreshed_at.sql diff --git a/cmd/agent/main.go b/cmd/agent/main.go index d4e1e0b..96aedff 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -296,6 +296,9 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S } go d.handleTreeList(ctx, env.ID, p, tx) + case api.MsgSnapshotsRefresh: + go d.refreshSnapshots(ctx, tx) + case api.MsgScheduleSet: var p api.ScheduleSetPayload if err := env.UnmarshalPayload(&p); err != nil { @@ -405,6 +408,22 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S return nil } +func (d *dispatcher) refreshSnapshots(ctx context.Context, tx wsclient.Sender) { + creds, err := d.secrets.Load() + if err != nil || creds.Empty() { + slog.Warn("ws agent: snapshots.refresh unavailable", "err", err) + return + } + r := runner.New(runner.Config{ + ResticBin: d.resticBin, ResticVersion: d.resticVer, + RepoURL: creds.URL, RepoUsername: creds.Username, RepoPassword: creds.Password, + SupportsRestoreNoOwnership: d.resticSupportsNoOwnership, + }, tx, time.Second) + if err := r.RefreshSnapshots(ctx); err != nil { + slog.Warn("ws agent: snapshots.refresh failed", "err", err) + } +} + // handleTreeList runs `restic ls --json ` and ships // the matching tree.list.result envelope back, correlated by the // request envelope's ID. Errors (missing creds, restic failure) diff --git a/internal/agent/runner/runner.go b/internal/agent/runner/runner.go index 9333e1d..ccfba23 100644 --- a/internal/agent/runner/runner.go +++ b/internal/agent/runner/runner.go @@ -77,6 +77,12 @@ func (r *Runner) resticEnv() restic.Env { } } +// RefreshSnapshots reconciles the server's cached projection without running +// a mutating repository job. +func (r *Runner) RefreshSnapshots(ctx context.Context) error { + return r.reportSnapshots(ctx, r.resticEnv()) +} + // sendStarted ships a job.started envelope. func (r *Runner) sendStarted(jobID string, kind api.JobKind, startedAt time.Time) { env, _ := api.Marshal(api.MsgJobStarted, jobID, api.JobStartedPayload{ @@ -226,14 +232,9 @@ func (r *Runner) RunBackup(ctx context.Context, jobID string, paths, excludes, t } } - r.sendFinished(ctx, jobID, finishedAt, err, statsBlob) - // On a successful backup, refresh the server's snapshot projection. - // We do this *after* job.finished so the UI sees the job land first; - // the snapshot list is a follow-up that the host detail page polls - // or the dashboard sees on its next refresh. A failure here is - // logged but doesn't fail the job — the next successful backup will - // catch the projection up. + // Do this before job.finished so a failure in terminal reporting cannot + // prevent the independently useful projection refresh. if err == nil { if rerr := r.reportSnapshots(ctx, env); rerr != nil { slog.Warn("runner: snapshots.report failed", "job_id", jobID, "err", rerr) @@ -242,6 +243,7 @@ func (r *Runner) RunBackup(ctx context.Context, jobID string, paths, excludes, t slog.Warn("runner: stats.report after backup failed", "job_id", jobID, "err", rerr) } } + r.sendFinished(ctx, jobID, finishedAt, err, statsBlob) if err != nil { return fmt.Errorf("runner backup: %w", err) @@ -282,8 +284,6 @@ func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.Fo var seq atomic.Int64 err := env.RunForget(ctx, groups, dryRun, r.streamHandler(jobID, &seq)) finishedAt := time.Now().UTC() - r.sendFinished(ctx, jobID, finishedAt, err, nil) - // Refresh the server's snapshot projection — forget rewrites the // index so the host's snapshot list almost certainly shrunk. if err == nil { @@ -292,6 +292,7 @@ func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.Fo "job_id", jobID, "err", rerr) } } + r.sendFinished(ctx, jobID, finishedAt, err, nil) if err != nil { return fmt.Errorf("runner forget: %w", err) @@ -318,6 +319,9 @@ func (r *Runner) RunPrune(ctx context.Context, jobID string) error { if rerr := r.reportStats(ctx, env, api.RepoStatsPayload{LastPruneAt: &pruneAt}); rerr != nil { slog.Warn("runner: stats.report after prune failed", "job_id", jobID, "err", rerr) } + if rerr := r.reportSnapshots(ctx, env); rerr != nil { + slog.Warn("runner: snapshots.report after prune failed", "job_id", jobID, "err", rerr) + } } r.sendFinished(ctx, jobID, finishedAt, err, nil) diff --git a/internal/agent/runner/runner_test.go b/internal/agent/runner/runner_test.go index 956ac21..14e7b7b 100644 --- a/internal/agent/runner/runner_test.go +++ b/internal/agent/runner/runner_test.go @@ -116,7 +116,8 @@ func envelopeOrder(envs []api.Envelope) []api.MessageType { // TestRunPruneShipsExpectedEnvelopes drives RunPrune with a fake // binary that prints "prune" on stdout (for the log.stream envelope) // and emits valid stats JSON so reportStats can populate size fields. -// Expected sequence: job.started → log.stream → repo.stats → job.finished. +// Expected sequence: job.started → log.stream → repo.stats → snapshots.report +// → job.finished. func TestRunPruneShipsExpectedEnvelopes(t *testing.T) { t.Parallel() @@ -126,6 +127,7 @@ func TestRunPruneShipsExpectedEnvelopes(t *testing.T) { case "$1" in prune) echo "prune" ;; stats) echo '`+statsJSON+`' ;; + snapshots) echo "[]" ;; *) echo "unknown: $*" ;; esac `) @@ -138,7 +140,7 @@ esac order := envelopeOrder(tx.envs) // Confirm landmark envelope types appear in the required order. - wantTypes := []api.MessageType{api.MsgJobStarted, api.MsgLogStream, api.MsgRepoStats, api.MsgJobFinished} + wantTypes := []api.MessageType{api.MsgJobStarted, api.MsgLogStream, api.MsgRepoStats, api.MsgSnapshotsRpt, api.MsgJobFinished} positions := map[api.MessageType]int{} for i, mt := range order { if _, seen := positions[mt]; !seen { @@ -379,6 +381,15 @@ func TestRunInitShipsStartedAndFinished(t *testing.T) { _ = firstEnvOfType(t, tx.envs, api.MsgJobFinished) } +func firstIndexOfType(envs []api.Envelope, typ api.MessageType) int { + for i, env := range envs { + if env.Type == typ { + return i + } + } + return -1 +} + // TestRunForgetShipsStartedAndFinished confirms the refactored // RunForget still produces job.started and job.finished envelopes. func TestRunForgetShipsStartedAndFinished(t *testing.T) { @@ -402,5 +413,9 @@ esac t.Fatalf("RunForget: %v", err) } _ = firstEnvOfType(t, tx.envs, api.MsgJobStarted) - _ = firstEnvOfType(t, tx.envs, api.MsgJobFinished) + finished := firstIndexOfType(tx.envs, api.MsgJobFinished) + refreshed := firstIndexOfType(tx.envs, api.MsgSnapshotsRpt) + if refreshed < 0 || finished < 0 || refreshed >= finished { + t.Fatalf("snapshot refresh must precede terminal report: %v", envelopeOrder(tx.envs)) + } } diff --git a/internal/api/wire.go b/internal/api/wire.go index 1de58ae..fd6022e 100644 --- a/internal/api/wire.go +++ b/internal/api/wire.go @@ -36,12 +36,13 @@ const ( // Server → agent message types. const ( - MsgCommandRun MessageType = "command.run" - MsgCommandCancel MessageType = "command.cancel" - MsgScheduleSet MessageType = "schedule.set" - MsgConfigUpdate MessageType = "config.update" - MsgCommandUpdate MessageType = "command.update" - MsgTreeList MessageType = "tree.list" // sync RPC: list a snapshot's children + MsgCommandRun MessageType = "command.run" + MsgCommandCancel MessageType = "command.cancel" + MsgScheduleSet MessageType = "schedule.set" + MsgConfigUpdate MessageType = "config.update" + MsgCommandUpdate MessageType = "command.update" + MsgTreeList MessageType = "tree.list" // sync RPC: list a snapshot's children + MsgSnapshotsRefresh MessageType = "snapshots.refresh" ) // Envelope is the framing for every WS message in either direction. diff --git a/internal/server/http/server.go b/internal/server/http/server.go index 465b1b2..af63427 100644 --- a/internal/server/http/server.go +++ b/internal/server/http/server.go @@ -262,6 +262,7 @@ func (s *Server) routes(r chi.Router) { r.Post("/api/hosts/{id}/repo/unlock", s.handleRunRepoUnlock) r.Post("/api/jobs/{id}/cancel", s.handleCancelJob) r.Post("/api/hosts/{id}/snapshots/diff", s.handleSnapshotDiff) + r.Post("/api/hosts/{id}/snapshots/refresh", s.handleRefreshHostSnapshots) // HTMX form variants outside /api. r.Post("/hosts/{id}/snapshots/diff", s.handleSnapshotDiff) diff --git a/internal/server/http/snapshots.go b/internal/server/http/snapshots.go index d88e18b..48ecd0d 100644 --- a/internal/server/http/snapshots.go +++ b/internal/server/http/snapshots.go @@ -5,6 +5,10 @@ import ( "time" "github.com/go-chi/chi/v5" + "github.com/oklog/ulid/v2" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/api" + "gitea.dcglab.co.uk/steve/restic-manager/internal/store" ) // snapshotView is the public JSON shape for a snapshot. Matches the @@ -26,6 +30,7 @@ type listSnapshotsResponse struct { HostID string `json:"host_id"` Count int `json:"count"` RefreshedAt *time.Time `json:"refreshed_at,omitempty"` + Stale bool `json:"stale"` Snapshots []snapshotView `json:"snapshots"` } @@ -45,7 +50,8 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re return } - if _, err := s.deps.Store.GetHost(r.Context(), hostID); err != nil { + host, err := s.deps.Store.GetHost(r.Context(), hostID) + if err != nil { writeJSONError(w, stdhttp.StatusNotFound, "host_not_found", "") return } @@ -61,10 +67,13 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re Count: len(snaps), Snapshots: make([]snapshotView, len(snaps)), } - if len(snaps) > 0 { - t := snaps[0].RefreshedAt - out.RefreshedAt = &t + out.RefreshedAt = host.SnapshotRefreshedAt + mutationAt, err := s.deps.Store.LatestSuccessfulRepoMutation(r.Context(), hostID) + if err != nil { + writeJSONError(w, stdhttp.StatusInternalServerError, "internal", "") + return } + out.Stale = mutationAt != nil && (out.RefreshedAt == nil || out.RefreshedAt.Before(*mutationAt)) for i, sn := range snaps { out.Snapshots[i] = snapshotView{ ID: sn.ID, @@ -80,3 +89,31 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re writeJSON(w, stdhttp.StatusOK, out) } + +func (s *Server) handleRefreshHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Request) { + user, ok := s.requireUser(r) + if !ok { + writeJSONError(w, stdhttp.StatusUnauthorized, "unauthorised", "") + return + } + hostID := chi.URLParam(r, "id") + if _, err := s.deps.Store.GetHost(r.Context(), hostID); err != nil { + writeJSONError(w, stdhttp.StatusNotFound, "host_not_found", "") + return + } + if s.deps.Hub == nil || !s.deps.Hub.Connected(hostID) { + writeJSONError(w, stdhttp.StatusConflict, "host_offline", "agent is not currently connected") + return + } + env, _ := api.Marshal(api.MsgSnapshotsRefresh, ulid.Make().String(), nil) + if err := s.deps.Hub.Send(r.Context(), hostID, env); err != nil { + writeJSONError(w, stdhttp.StatusConflict, "host_offline", err.Error()) + return + } + now := time.Now().UTC() + _ = s.deps.Store.AppendAudit(r.Context(), store.AuditEntry{ + ID: ulid.Make().String(), UserID: &user.ID, Actor: "user", + Action: "host.snapshots_refresh", TargetKind: ptr("host"), TargetID: &hostID, TS: now, + }) + writeJSON(w, stdhttp.StatusAccepted, map[string]string{"status": "refresh_requested"}) +} diff --git a/internal/server/http/snapshots_test.go b/internal/server/http/snapshots_test.go new file mode 100644 index 0000000..d9fab37 --- /dev/null +++ b/internal/server/http/snapshots_test.go @@ -0,0 +1,83 @@ +package http + +import ( + "context" + "encoding/json" + stdhttp "net/http" + "testing" + "time" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/api" + "gitea.dcglab.co.uk/steve/restic-manager/internal/store" +) + +func TestSnapshotsFreshnessAndExplicitRefresh(t *testing.T) { + t.Parallel() + srv, ts, st := rawTestServerWithUI(t) + hostID, token := enrolHostForUI(t, srv, st, "snapshot-refresh-host") + c := agentDial(t, srv, ts, hostID, token) + sendHello(t, c, "snapshot-refresh-host") + _ = drainUntil(t, c, api.MsgScheduleSet) + cookie := loginAsAdmin(t, st) + + mutationAt := time.Now().UTC().Add(-time.Minute).Truncate(time.Millisecond) + if err := st.CreateJob(context.Background(), store.Job{ + ID: "mutation-job", HostID: hostID, Kind: "forget", ActorKind: "user", CreatedAt: mutationAt.Add(-time.Minute), + }); err != nil { + t.Fatalf("create mutation: %v", err) + } + if err := st.MarkJobFinished(context.Background(), "mutation-job", "succeeded", 0, nil, "", mutationAt); err != nil { + t.Fatalf("finish mutation: %v", err) + } + + get := func() listSnapshotsResponse { + req, _ := stdhttp.NewRequest(stdhttp.MethodGet, ts.URL+"/api/hosts/"+hostID+"/snapshots", nil) + req.AddCookie(cookie) + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + t.Fatalf("get snapshots: %v", err) + } + defer res.Body.Close() + var body listSnapshotsResponse + if err := json.NewDecoder(res.Body).Decode(&body); err != nil { + t.Fatalf("decode snapshots: %v", err) + } + return body + } + if body := get(); !body.Stale || body.RefreshedAt != nil { + t.Fatalf("unrefreshed projection should be stale: %+v", body) + } + + refreshedAt := mutationAt.Add(time.Second) + if err := st.ReplaceHostSnapshots(context.Background(), hostID, nil, refreshedAt); err != nil { + t.Fatalf("replace empty: %v", err) + } + if body := get(); body.Stale || body.RefreshedAt == nil || !body.RefreshedAt.Equal(refreshedAt) { + t.Fatalf("fresh empty projection reported incorrectly: %+v", body) + } + + req, _ := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/hosts/"+hostID+"/snapshots/refresh", nil) + req.AddCookie(cookie) + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + t.Fatalf("request refresh: %v", err) + } + defer res.Body.Close() + if res.StatusCode != stdhttp.StatusAccepted { + t.Fatalf("refresh status = %d, want 202", res.StatusCode) + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _, raw, err := c.Read(ctx) + if err != nil { + t.Fatalf("read refresh command: %v", err) + } + var env api.Envelope + if err := json.Unmarshal(raw, &env); err != nil { + t.Fatalf("decode envelope: %v", err) + } + if env.Type != api.MsgSnapshotsRefresh { + t.Fatalf("message type = %q, want %q", env.Type, api.MsgSnapshotsRefresh) + } +} diff --git a/internal/store/hosts.go b/internal/store/hosts.go index 9afa5c7..7d4f225 100644 --- a/internal/store/hosts.go +++ b/internal/store/hosts.go @@ -44,7 +44,7 @@ func (s *Store) LookupHostByAgentToken(ctx context.Context, tokenHash string) (* repo_size_bytes, snapshot_count, open_alert_count, applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps, pre_hook_default, post_hook_default, - repo_status, repo_status_error, always_on + repo_status, repo_status_error, always_on, snapshot_refreshed_at FROM hosts WHERE agent_token_hash = ?`, tokenHash) return scanHost(row) @@ -59,7 +59,7 @@ func (s *Store) GetHost(ctx context.Context, id string) (*Host, error) { repo_size_bytes, snapshot_count, open_alert_count, applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps, pre_hook_default, post_hook_default, - repo_status, repo_status_error, always_on + repo_status, repo_status_error, always_on, snapshot_refreshed_at FROM hosts WHERE id = ?`, id) return scanHost(row) } @@ -227,7 +227,7 @@ func (s *Store) ListHosts(ctx context.Context) ([]Host, error) { repo_size_bytes, snapshot_count, open_alert_count, applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps, pre_hook_default, post_hook_default, - repo_status, repo_status_error, always_on + repo_status, repo_status_error, always_on, snapshot_refreshed_at FROM hosts ORDER BY name`) if err != nil { return nil, fmt.Errorf("store: list hosts: %w", err) @@ -268,6 +268,7 @@ func scanHostRow(s hostScanner) (*Host, error) { bwUp, bwDown sql.NullInt64 preHook, postHook sql.NullString alwaysOn int + snapshotRefreshedAt sql.NullString ) err := s.Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.AgentVersion, &h.ResticVersion, &h.ProtocolVersion, @@ -276,7 +277,7 @@ func scanHostRow(s hostScanner) (*Host, error) { &h.RepoSizeBytes, &h.SnapshotCount, &h.OpenAlertCount, &h.AppliedScheduleVersion, &bwUp, &bwDown, &preHook, &postHook, - &h.RepoStatus, &h.RepoStatusError, &alwaysOn) + &h.RepoStatus, &h.RepoStatusError, &alwaysOn, &snapshotRefreshedAt) if err != nil { if errors.Is(err, sql.ErrNoRows) { return nil, ErrNotFound @@ -332,6 +333,13 @@ func scanHostRow(s hostScanner) (*Host, error) { h.PostHookDefault = postHook.String } h.AlwaysOn = alwaysOn != 0 + if snapshotRefreshedAt.Valid { + t, err := time.Parse(time.RFC3339Nano, snapshotRefreshedAt.String) + if err != nil { + return nil, fmt.Errorf("store: parse snapshot_refreshed_at: %w", err) + } + h.SnapshotRefreshedAt = &t + } return &h, nil } diff --git a/internal/store/jobs.go b/internal/store/jobs.go index cd1ff8f..02833b8 100644 --- a/internal/store/jobs.go +++ b/internal/store/jobs.go @@ -39,6 +39,27 @@ type RunningJobActivity struct { LastActivity time.Time } +// LatestSuccessfulRepoMutation returns the newest completion time for a job +// that can change the repository's snapshot projection. +func (s *Store) LatestSuccessfulRepoMutation(ctx context.Context, hostID string) (*time.Time, error) { + var raw sql.NullString + err := s.db.QueryRowContext(ctx, ` + SELECT MAX(finished_at) FROM jobs + WHERE host_id = ? AND status = 'succeeded' + AND kind IN ('backup', 'forget', 'prune')`, hostID).Scan(&raw) + if err != nil { + return nil, fmt.Errorf("store: latest successful repo mutation: %w", err) + } + if !raw.Valid { + return nil, nil + } + t, err := time.Parse(time.RFC3339Nano, raw.String) + if err != nil { + return nil, fmt.Errorf("store: parse latest repo mutation: %w", err) + } + return &t, nil +} + // ListRunningJobActivity returns every running job with its latest persisted // activity. Jobs without started_at are excluded because they have not actually // entered the running state coherently and cannot be aged safely here. diff --git a/internal/store/migrations/0025_hosts_snapshot_refreshed_at.sql b/internal/store/migrations/0025_hosts_snapshot_refreshed_at.sql new file mode 100644 index 0000000..f7defe6 --- /dev/null +++ b/internal/store/migrations/0025_hosts_snapshot_refreshed_at.sql @@ -0,0 +1 @@ +ALTER TABLE hosts ADD COLUMN snapshot_refreshed_at TEXT; diff --git a/internal/store/snapshots.go b/internal/store/snapshots.go index 9995830..cfbda58 100644 --- a/internal/store/snapshots.go +++ b/internal/store/snapshots.go @@ -69,8 +69,8 @@ func (s *Store) ReplaceHostSnapshots(ctx context.Context, hostID string, snaps [ } if _, err := tx.ExecContext(ctx, - `UPDATE hosts SET snapshot_count = ? WHERE id = ?`, - len(snaps), hostID); err != nil { + `UPDATE hosts SET snapshot_count = ?, snapshot_refreshed_at = ? WHERE id = ?`, + len(snaps), when.UTC().Format(time.RFC3339Nano), hostID); err != nil { return fmt.Errorf("store: update host snapshot_count: %w", err) } diff --git a/internal/store/snapshots_test.go b/internal/store/snapshots_test.go index 2abc9a3..b83f42b 100644 --- a/internal/store/snapshots_test.go +++ b/internal/store/snapshots_test.go @@ -138,7 +138,8 @@ func TestReplaceHostSnapshotsEmpty(t *testing.T) { t.Fatalf("replace 1: %v", err) } // Then empty — host has been wiped. - if err := s.ReplaceHostSnapshots(ctx, hostID, nil, time.Now().UTC()); err != nil { + refreshedAt := time.Now().UTC().Truncate(time.Millisecond) + if err := s.ReplaceHostSnapshots(ctx, hostID, nil, refreshedAt); err != nil { t.Fatalf("replace empty: %v", err) } out, err := s.ListSnapshotsByHost(ctx, hostID) @@ -152,4 +153,7 @@ func TestReplaceHostSnapshotsEmpty(t *testing.T) { if h.SnapshotCount != 0 { t.Errorf("snapshot_count should reset to 0, got %d", h.SnapshotCount) } + if h.SnapshotRefreshedAt == nil || !h.SnapshotRefreshedAt.Equal(refreshedAt) { + t.Errorf("empty projection refresh time = %v, want %s", h.SnapshotRefreshedAt, refreshedAt) + } } diff --git a/internal/store/types.go b/internal/store/types.go index 9ad6326..a467557 100644 --- a/internal/store/types.go +++ b/internal/store/types.go @@ -77,6 +77,7 @@ type Host struct { LastBackupStatus *string RepoSizeBytes int64 SnapshotCount int + SnapshotRefreshedAt *time.Time OpenAlertCount int AppliedScheduleVersion int64 // Host-wide bandwidth caps applied to every restic invocation -- 2.52.0