Make snapshot projections fresh and self-diagnosing #47

Merged
steve merged 1 commits from fix-issue-40-snapshot-reconciliation into main 2026-08-22 11:03:47 +01:00
13 changed files with 224 additions and 29 deletions
+19
View File
@@ -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) go d.handleTreeList(ctx, env.ID, p, tx)
case api.MsgSnapshotsRefresh:
go d.refreshSnapshots(ctx, tx)
case api.MsgScheduleSet: case api.MsgScheduleSet:
var p api.ScheduleSetPayload var p api.ScheduleSetPayload
if err := env.UnmarshalPayload(&p); err != nil { 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 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 <snapshot> <path>` and ships // handleTreeList runs `restic ls --json <snapshot> <path>` and ships
// the matching tree.list.result envelope back, correlated by the // the matching tree.list.result envelope back, correlated by the
// request envelope's ID. Errors (missing creds, restic failure) // request envelope's ID. Errors (missing creds, restic failure)
+13 -9
View File
@@ -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. // sendStarted ships a job.started envelope.
func (r *Runner) sendStarted(jobID string, kind api.JobKind, startedAt time.Time) { func (r *Runner) sendStarted(jobID string, kind api.JobKind, startedAt time.Time) {
env, _ := api.Marshal(api.MsgJobStarted, jobID, api.JobStartedPayload{ 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. // On a successful backup, refresh the server's snapshot projection.
// We do this *after* job.finished so the UI sees the job land first; // Do this before job.finished so a failure in terminal reporting cannot
// the snapshot list is a follow-up that the host detail page polls // prevent the independently useful projection refresh.
// 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.
if err == nil { if err == nil {
if rerr := r.reportSnapshots(ctx, env); rerr != nil { if rerr := r.reportSnapshots(ctx, env); rerr != nil {
slog.Warn("runner: snapshots.report failed", "job_id", jobID, "err", rerr) 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) slog.Warn("runner: stats.report after backup failed", "job_id", jobID, "err", rerr)
} }
} }
r.sendFinished(ctx, jobID, finishedAt, err, statsBlob)
if err != nil { if err != nil {
return fmt.Errorf("runner backup: %w", err) 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 var seq atomic.Int64
err := env.RunForget(ctx, groups, dryRun, r.streamHandler(jobID, &seq)) err := env.RunForget(ctx, groups, dryRun, r.streamHandler(jobID, &seq))
finishedAt := time.Now().UTC() finishedAt := time.Now().UTC()
r.sendFinished(ctx, jobID, finishedAt, err, nil)
// Refresh the server's snapshot projection — forget rewrites the // Refresh the server's snapshot projection — forget rewrites the
// index so the host's snapshot list almost certainly shrunk. // index so the host's snapshot list almost certainly shrunk.
if err == nil { if err == nil {
@@ -292,6 +292,7 @@ func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.Fo
"job_id", jobID, "err", rerr) "job_id", jobID, "err", rerr)
} }
} }
r.sendFinished(ctx, jobID, finishedAt, err, nil)
if err != nil { if err != nil {
return fmt.Errorf("runner forget: %w", err) 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 { 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) 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) r.sendFinished(ctx, jobID, finishedAt, err, nil)
+18 -3
View File
@@ -116,7 +116,8 @@ func envelopeOrder(envs []api.Envelope) []api.MessageType {
// TestRunPruneShipsExpectedEnvelopes drives RunPrune with a fake // TestRunPruneShipsExpectedEnvelopes drives RunPrune with a fake
// binary that prints "prune" on stdout (for the log.stream envelope) // binary that prints "prune" on stdout (for the log.stream envelope)
// and emits valid stats JSON so reportStats can populate size fields. // 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) { func TestRunPruneShipsExpectedEnvelopes(t *testing.T) {
t.Parallel() t.Parallel()
@@ -126,6 +127,7 @@ func TestRunPruneShipsExpectedEnvelopes(t *testing.T) {
case "$1" in case "$1" in
prune) echo "prune" ;; prune) echo "prune" ;;
stats) echo '`+statsJSON+`' ;; stats) echo '`+statsJSON+`' ;;
snapshots) echo "[]" ;;
*) echo "unknown: $*" ;; *) echo "unknown: $*" ;;
esac esac
`) `)
@@ -138,7 +140,7 @@ esac
order := envelopeOrder(tx.envs) order := envelopeOrder(tx.envs)
// Confirm landmark envelope types appear in the required order. // 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{} positions := map[api.MessageType]int{}
for i, mt := range order { for i, mt := range order {
if _, seen := positions[mt]; !seen { if _, seen := positions[mt]; !seen {
@@ -379,6 +381,15 @@ func TestRunInitShipsStartedAndFinished(t *testing.T) {
_ = firstEnvOfType(t, tx.envs, api.MsgJobFinished) _ = 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 // TestRunForgetShipsStartedAndFinished confirms the refactored
// RunForget still produces job.started and job.finished envelopes. // RunForget still produces job.started and job.finished envelopes.
func TestRunForgetShipsStartedAndFinished(t *testing.T) { func TestRunForgetShipsStartedAndFinished(t *testing.T) {
@@ -402,5 +413,9 @@ esac
t.Fatalf("RunForget: %v", err) t.Fatalf("RunForget: %v", err)
} }
_ = firstEnvOfType(t, tx.envs, api.MsgJobStarted) _ = 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))
}
} }
+1
View File
@@ -42,6 +42,7 @@ const (
MsgConfigUpdate MessageType = "config.update" MsgConfigUpdate MessageType = "config.update"
MsgCommandUpdate MessageType = "command.update" MsgCommandUpdate MessageType = "command.update"
MsgTreeList MessageType = "tree.list" // sync RPC: list a snapshot's children 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. // Envelope is the framing for every WS message in either direction.
+1
View File
@@ -262,6 +262,7 @@ func (s *Server) routes(r chi.Router) {
r.Post("/api/hosts/{id}/repo/unlock", s.handleRunRepoUnlock) r.Post("/api/hosts/{id}/repo/unlock", s.handleRunRepoUnlock)
r.Post("/api/jobs/{id}/cancel", s.handleCancelJob) r.Post("/api/jobs/{id}/cancel", s.handleCancelJob)
r.Post("/api/hosts/{id}/snapshots/diff", s.handleSnapshotDiff) r.Post("/api/hosts/{id}/snapshots/diff", s.handleSnapshotDiff)
r.Post("/api/hosts/{id}/snapshots/refresh", s.handleRefreshHostSnapshots)
// HTMX form variants outside /api. // HTMX form variants outside /api.
r.Post("/hosts/{id}/snapshots/diff", s.handleSnapshotDiff) r.Post("/hosts/{id}/snapshots/diff", s.handleSnapshotDiff)
+41 -4
View File
@@ -5,6 +5,10 @@ import (
"time" "time"
"github.com/go-chi/chi/v5" "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 // snapshotView is the public JSON shape for a snapshot. Matches the
@@ -26,6 +30,7 @@ type listSnapshotsResponse struct {
HostID string `json:"host_id"` HostID string `json:"host_id"`
Count int `json:"count"` Count int `json:"count"`
RefreshedAt *time.Time `json:"refreshed_at,omitempty"` RefreshedAt *time.Time `json:"refreshed_at,omitempty"`
Stale bool `json:"stale"`
Snapshots []snapshotView `json:"snapshots"` Snapshots []snapshotView `json:"snapshots"`
} }
@@ -45,7 +50,8 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re
return 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", "") writeJSONError(w, stdhttp.StatusNotFound, "host_not_found", "")
return return
} }
@@ -61,10 +67,13 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re
Count: len(snaps), Count: len(snaps),
Snapshots: make([]snapshotView, len(snaps)), Snapshots: make([]snapshotView, len(snaps)),
} }
if len(snaps) > 0 { out.RefreshedAt = host.SnapshotRefreshedAt
t := snaps[0].RefreshedAt mutationAt, err := s.deps.Store.LatestSuccessfulRepoMutation(r.Context(), hostID)
out.RefreshedAt = &t 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 { for i, sn := range snaps {
out.Snapshots[i] = snapshotView{ out.Snapshots[i] = snapshotView{
ID: sn.ID, ID: sn.ID,
@@ -80,3 +89,31 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re
writeJSON(w, stdhttp.StatusOK, out) 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"})
}
+83
View File
@@ -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)
}
}
+12 -4
View File
@@ -44,7 +44,7 @@ func (s *Store) LookupHostByAgentToken(ctx context.Context, tokenHash string) (*
repo_size_bytes, snapshot_count, open_alert_count, repo_size_bytes, snapshot_count, open_alert_count,
applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps, applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps,
pre_hook_default, post_hook_default, 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 = ?`, FROM hosts WHERE agent_token_hash = ?`,
tokenHash) tokenHash)
return scanHost(row) 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, repo_size_bytes, snapshot_count, open_alert_count,
applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps, applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps,
pre_hook_default, post_hook_default, 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) FROM hosts WHERE id = ?`, id)
return scanHost(row) return scanHost(row)
} }
@@ -227,7 +227,7 @@ func (s *Store) ListHosts(ctx context.Context) ([]Host, error) {
repo_size_bytes, snapshot_count, open_alert_count, repo_size_bytes, snapshot_count, open_alert_count,
applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps, applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps,
pre_hook_default, post_hook_default, 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`) FROM hosts ORDER BY name`)
if err != nil { if err != nil {
return nil, fmt.Errorf("store: list hosts: %w", err) return nil, fmt.Errorf("store: list hosts: %w", err)
@@ -268,6 +268,7 @@ func scanHostRow(s hostScanner) (*Host, error) {
bwUp, bwDown sql.NullInt64 bwUp, bwDown sql.NullInt64
preHook, postHook sql.NullString preHook, postHook sql.NullString
alwaysOn int alwaysOn int
snapshotRefreshedAt sql.NullString
) )
err := s.Scan(&h.ID, &h.Name, &h.OS, &h.Arch, err := s.Scan(&h.ID, &h.Name, &h.OS, &h.Arch,
&h.AgentVersion, &h.ResticVersion, &h.ProtocolVersion, &h.AgentVersion, &h.ResticVersion, &h.ProtocolVersion,
@@ -276,7 +277,7 @@ func scanHostRow(s hostScanner) (*Host, error) {
&h.RepoSizeBytes, &h.SnapshotCount, &h.OpenAlertCount, &h.RepoSizeBytes, &h.SnapshotCount, &h.OpenAlertCount,
&h.AppliedScheduleVersion, &bwUp, &bwDown, &h.AppliedScheduleVersion, &bwUp, &bwDown,
&preHook, &postHook, &preHook, &postHook,
&h.RepoStatus, &h.RepoStatusError, &alwaysOn) &h.RepoStatus, &h.RepoStatusError, &alwaysOn, &snapshotRefreshedAt)
if err != nil { if err != nil {
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return nil, ErrNotFound return nil, ErrNotFound
@@ -332,6 +333,13 @@ func scanHostRow(s hostScanner) (*Host, error) {
h.PostHookDefault = postHook.String h.PostHookDefault = postHook.String
} }
h.AlwaysOn = alwaysOn != 0 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 return &h, nil
} }
+21
View File
@@ -39,6 +39,27 @@ type RunningJobActivity struct {
LastActivity time.Time 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 // ListRunningJobActivity returns every running job with its latest persisted
// activity. Jobs without started_at are excluded because they have not actually // activity. Jobs without started_at are excluded because they have not actually
// entered the running state coherently and cannot be aged safely here. // entered the running state coherently and cannot be aged safely here.
@@ -0,0 +1 @@
ALTER TABLE hosts ADD COLUMN snapshot_refreshed_at TEXT;
+2 -2
View File
@@ -69,8 +69,8 @@ func (s *Store) ReplaceHostSnapshots(ctx context.Context, hostID string, snaps [
} }
if _, err := tx.ExecContext(ctx, if _, err := tx.ExecContext(ctx,
`UPDATE hosts SET snapshot_count = ? WHERE id = ?`, `UPDATE hosts SET snapshot_count = ?, snapshot_refreshed_at = ? WHERE id = ?`,
len(snaps), hostID); err != nil { len(snaps), when.UTC().Format(time.RFC3339Nano), hostID); err != nil {
return fmt.Errorf("store: update host snapshot_count: %w", err) return fmt.Errorf("store: update host snapshot_count: %w", err)
} }
+5 -1
View File
@@ -138,7 +138,8 @@ func TestReplaceHostSnapshotsEmpty(t *testing.T) {
t.Fatalf("replace 1: %v", err) t.Fatalf("replace 1: %v", err)
} }
// Then empty — host has been wiped. // 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) t.Fatalf("replace empty: %v", err)
} }
out, err := s.ListSnapshotsByHost(ctx, hostID) out, err := s.ListSnapshotsByHost(ctx, hostID)
@@ -152,4 +153,7 @@ func TestReplaceHostSnapshotsEmpty(t *testing.T) {
if h.SnapshotCount != 0 { if h.SnapshotCount != 0 {
t.Errorf("snapshot_count should reset to 0, got %d", h.SnapshotCount) 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)
}
} }
+1
View File
@@ -77,6 +77,7 @@ type Host struct {
LastBackupStatus *string LastBackupStatus *string
RepoSizeBytes int64 RepoSizeBytes int64
SnapshotCount int SnapshotCount int
SnapshotRefreshedAt *time.Time
OpenAlertCount int OpenAlertCount int
AppliedScheduleVersion int64 AppliedScheduleVersion int64
// Host-wide bandwidth caps applied to every restic invocation // Host-wide bandwidth caps applied to every restic invocation