Make snapshot projections self-diagnosing
CI / Test (rest) (pull_request) Successful in 40s
CI / Test (store) (pull_request) Successful in 42s
CI / Lint (pull_request) Successful in 11s
CI / Build (windows/amd64) (pull_request) Successful in 8s
CI / Build (linux/arm64) (pull_request) Successful in 7s
CI / Build (linux/amd64) (pull_request) Successful in 8s
CI / Test (server-http) (pull_request) Successful in 1m32s
e2e / Playwright vs docker-compose (pull_request) Successful in 1m26s

This commit is contained in:
2026-08-22 11:00:46 +01:00
parent 8fdf4a1bdf
commit f9718e6077
13 changed files with 224 additions and 29 deletions
+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.
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)
+18 -3
View File
@@ -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))
}
}