From cdf88c6dc384e50b45bb28cd94b4d4ab5543f8d1 Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Mon, 4 May 2026 10:38:34 +0100 Subject: [PATCH 01/15] agent+server: apply host bandwidth caps to restic invocations P2R-13a. restic.Env gains LimitUploadKBps/LimitDownloadKBps which are emitted as global --limit-upload/--limit-download flags before the subcommand on every invocation. Agent dispatcher tracks host-wide caps received via config.update; server pushes them on hello and after PUT /api/hosts/{id}/bandwidth. Also extends api.CommandRunPayload with optional per-job overrides (BandwidthUpKBps/Down + PreHook/PostHook); the override consumers land in T2/T6. --- cmd/agent/main.go | 59 +++++++++++++--- internal/agent/runner/runner.go | 16 +++-- internal/api/messages.go | 21 ++++++ internal/restic/runner.go | 61 ++++++++++------ internal/restic/runner_test.go | 37 ++++++++++ internal/server/http/host_bandwidth.go | 5 ++ internal/server/http/host_bandwidth_push.go | 78 +++++++++++++++++++++ internal/server/http/host_credentials.go | 4 ++ 8 files changed, 246 insertions(+), 35 deletions(-) create mode 100644 internal/server/http/host_bandwidth_push.go diff --git a/cmd/agent/main.go b/cmd/agent/main.go index d401640..d6c3d8b 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -9,6 +9,7 @@ import ( "os" "os/signal" "strconv" + "sync" "syscall" "time" @@ -170,6 +171,14 @@ type dispatcher struct { resticBin string secrets *secrets.Store scheduler *scheduler.Scheduler + + // Bandwidth caps in KB/s pushed via config.update. Mutated under + // bwMu by the config.update handler; read by runJob when building + // the runner. <=0 means "no cap" (do not pass --limit-* to restic). + // Per-job overrides on CommandRunPayload take precedence. + bwMu sync.Mutex + bwUpKBps int + bwDownKBps int } func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.Sender) error { @@ -263,6 +272,24 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S slog.Warn("ws agent: unknown config.update slot, ignoring", "slot", p.Slot) } + // Bandwidth caps ride independently of the slot — they're host- + // wide and apply to every restic invocation regardless of which + // credentials slot the job uses. nil pointer = no change in this + // push; non-nil = set to that value (≤0 clears the cap). + if p.BandwidthUpKBps != nil || p.BandwidthDownKBps != nil { + d.bwMu.Lock() + if p.BandwidthUpKBps != nil { + d.bwUpKBps = *p.BandwidthUpKBps + } + if p.BandwidthDownKBps != nil { + d.bwDownKBps = *p.BandwidthDownKBps + } + up, down := d.bwUpKBps, d.bwDownKBps + d.bwMu.Unlock() + slog.Info("ws agent: bandwidth caps updated", + "up_kbps", up, "down_kbps", down) + } + case api.MsgAgentUpdateAvail: var p api.AgentUpdateAvailablePayload _ = env.UnmarshalPayload(&p) @@ -295,11 +322,25 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc // not on r). If you find yourself adding a new JobKind that // needs delete authority, mirror the JobPrune pattern below // — don't try to overload r. + // Resolve bandwidth caps: per-job override (if set) wins over the + // host-wide caps last pushed via config.update. <=0 means no cap. + d.bwMu.Lock() + upKBps, downKBps := d.bwUpKBps, d.bwDownKBps + d.bwMu.Unlock() + if p.BandwidthUpKBps != nil { + upKBps = *p.BandwidthUpKBps + } + if p.BandwidthDownKBps != nil { + downKBps = *p.BandwidthDownKBps + } + r := runner.New(runner.Config{ - ResticBin: d.resticBin, - RepoURL: creds.URL, - RepoUsername: creds.Username, - RepoPassword: creds.Password, + ResticBin: d.resticBin, + RepoURL: creds.URL, + RepoUsername: creds.Username, + RepoPassword: creds.Password, + LimitUploadKBps: upKBps, + LimitDownloadKBps: downKBps, }, tx, time.Second) switch p.Kind { @@ -381,10 +422,12 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc runCreds = ac } prr := runner.New(runner.Config{ - ResticBin: d.resticBin, - RepoURL: runCreds.URL, - RepoUsername: runCreds.Username, - RepoPassword: runCreds.Password, + ResticBin: d.resticBin, + RepoURL: runCreds.URL, + RepoUsername: runCreds.Username, + RepoPassword: runCreds.Password, + LimitUploadKBps: upKBps, + LimitDownloadKBps: downKBps, }, tx, time.Second) slog.Info("agent: accepting prune job", "job_id", p.JobID, "admin_creds", p.RequiresAdminCreds) go func() { diff --git a/internal/agent/runner/runner.go b/internal/agent/runner/runner.go index 985380e..617c8a6 100644 --- a/internal/agent/runner/runner.go +++ b/internal/agent/runner/runner.go @@ -30,6 +30,12 @@ type Config struct { RepoURL string RepoUsername string RepoPassword string + + // Bandwidth caps in KB/s applied to every restic invocation. + // <=0 means "no cap". Per-job override: callers that build a + // runner per-dispatch can pass the override value here directly. + LimitUploadKBps int + LimitDownloadKBps int } // Runner owns the restic invocations. @@ -54,10 +60,12 @@ func New(cfg Config, tx Sender, progressMinPeriod time.Duration) *Runner { // resticEnv builds the shared restic.Env from r.cfg. func (r *Runner) resticEnv() restic.Env { return restic.Env{ - Bin: r.cfg.ResticBin, - RepoURL: r.cfg.RepoURL, - RepoUsername: r.cfg.RepoUsername, - RepoPassword: r.cfg.RepoPassword, + Bin: r.cfg.ResticBin, + RepoURL: r.cfg.RepoURL, + RepoUsername: r.cfg.RepoUsername, + RepoPassword: r.cfg.RepoPassword, + LimitUploadKBps: r.cfg.LimitUploadKBps, + LimitDownloadKBps: r.cfg.LimitDownloadKBps, } } diff --git a/internal/api/messages.go b/internal/api/messages.go index 816d203..ce43bc3 100644 --- a/internal/api/messages.go +++ b/internal/api/messages.go @@ -130,6 +130,19 @@ type CommandRunPayload struct { Tag string `json:"tag,omitempty"` ForgetGroups []ForgetGroup `json:"forget_groups,omitempty"` RequiresAdminCreds bool `json:"requires_admin_creds,omitempty"` + + // Per-job bandwidth caps in KB/s. When nil, the agent uses the + // host-wide caps it received via config.update. When non-nil, + // the override wins for this job only — even a non-nil zero + // pointer means "no cap for this job" (caller's explicit choice). + BandwidthUpKBps *int `json:"bandwidth_up_kbps,omitempty"` + BandwidthDownKBps *int `json:"bandwidth_down_kbps,omitempty"` + + // Hooks run only for kind=backup. Server resolves source-group + // hook → host default → empty before dispatching, so the agent + // just executes whatever is here. + PreHook string `json:"pre_hook,omitempty"` + PostHook string `json:"post_hook,omitempty"` } // CommandCancelPayload is the server → agent cancel signal. @@ -306,6 +319,14 @@ type ConfigUpdatePayload struct { RepoCredential string `json:"repo_credential,omitempty"` // sensitive (for rest server basic auth) HookShell string `json:"hook_shell,omitempty"` Slot string `json:"slot,omitempty"` + + // Bandwidth caps in KB/s. Pointer semantics so the server can + // disambiguate "no change in this push" (nil → omitted on the + // wire) from "explicitly clear the cap" (zero or negative value). + // Applied to every restic invocation as --limit-upload / + // --limit-download. Per-job overrides ride on CommandRunPayload. + BandwidthUpKBps *int `json:"bandwidth_up_kbps,omitempty"` + BandwidthDownKBps *int `json:"bandwidth_down_kbps,omitempty"` } // AgentUpdateAvailablePayload — informational only; the agent does diff --git a/internal/restic/runner.go b/internal/restic/runner.go index 51675ff..6104e7a 100644 --- a/internal/restic/runner.go +++ b/internal/restic/runner.go @@ -47,6 +47,37 @@ type Env struct { RepoPassword string // doubles as RESTIC_PASSWORD and (for rest:) HTTP basic-auth password ExtraEnv map[string]string // any other RESTIC_* / passthrough WorkDir string // CWD; default = current + + // Bandwidth caps in KB/s. <=0 means "no cap" (omit the flag). + // Emitted as restic global flags --limit-upload / --limit-download + // before the subcommand on every invocation. + LimitUploadKBps int + LimitDownloadKBps int +} + +// globalArgs returns restic's pre-subcommand global flags derived +// from the Env. Currently just bandwidth caps. +func (e Env) globalArgs() []string { + var out []string + if e.LimitUploadKBps > 0 { + out = append(out, "--limit-upload", fmt.Sprintf("%d", e.LimitUploadKBps)) + } + if e.LimitDownloadKBps > 0 { + out = append(out, "--limit-download", fmt.Sprintf("%d", e.LimitDownloadKBps)) + } + return out +} + +// resticCmd builds an exec.Cmd with bandwidth-limit globals prefixed +// before the supplied subcommand args. Centralizing this so every +// command (backup/forget/prune/check/unlock/init/stats) honors +// the caps without each call site having to remember. +func (e Env) resticCmd(ctx context.Context, sub ...string) *exec.Cmd { + args := append(e.globalArgs(), sub...) + cmd := exec.CommandContext(ctx, e.Bin, args...) + cmd.Env = e.envSlice() + cmd.Dir = e.WorkDir + return cmd } // EventKind enumerates what we care about in restic's --json output @@ -110,9 +141,7 @@ func (e Env) RunBackup(ctx context.Context, paths, excludes, tags []string, hand } args = append(args, paths...) - cmd := exec.CommandContext(ctx, e.Bin, args...) - cmd.Env = e.envSlice() - cmd.Dir = e.WorkDir + cmd := e.resticCmd(ctx, args...) stdout, err := cmd.StdoutPipe() if err != nil { @@ -215,9 +244,7 @@ func (e Env) RunForget(ctx context.Context, groups []ForgetGroup, handle LineHan } args := []string{"forget", "--json", "--tag", g.Tag} args = append(args, g.Policy.args()...) - cmd := exec.CommandContext(ctx, e.Bin, args...) - cmd.Env = e.envSlice() - cmd.Dir = e.WorkDir + cmd := e.resticCmd(ctx, args...) if err := runWithPump(cmd, handle); err != nil { return err } @@ -232,9 +259,7 @@ func (e Env) RunForget(ctx context.Context, groups []ForgetGroup, handle LineHan // at " on success, "config file already exists" on a // re-init attempt, etc.). func (e Env) RunInit(ctx context.Context, handle LineHandler) error { - cmd := exec.CommandContext(ctx, e.Bin, "init") - cmd.Env = e.envSlice() - cmd.Dir = e.WorkDir + cmd := e.resticCmd(ctx, "init") // Sniff for "config file already exists" on stderr; if we see it // we'll treat the non-zero exit as a soft success — running init @@ -272,10 +297,7 @@ func (e Env) RunInit(ctx context.Context, handle LineHandler) error { // support that's useful for our purposes). We tee everything to the // handler so the live log is the operator's progress bar. func (e Env) RunPrune(ctx context.Context, handle LineHandler) error { - cmd := exec.CommandContext(ctx, e.Bin, "prune") - cmd.Env = e.envSlice() - cmd.Dir = e.WorkDir - return runWithPump(cmd, handle) + return runWithPump(e.resticCmd(ctx, "prune"), handle) } // runWithPump starts the configured cmd, fans stdout+stderr into @@ -313,10 +335,7 @@ func runWithPump(cmd *exec.Cmd, handle LineHandler) error { // RunUnlock executes `restic unlock`. Returns nil on a clean exit. func (e Env) RunUnlock(ctx context.Context, handle LineHandler) error { - cmd := exec.CommandContext(ctx, e.Bin, "unlock") - cmd.Env = e.envSlice() - cmd.Dir = e.WorkDir - return runWithPump(cmd, handle) + return runWithPump(e.resticCmd(ctx, "unlock"), handle) } // RepoStats mirrors `restic stats --json --mode raw-data` output. @@ -333,9 +352,7 @@ type RepoStats struct { // caller can still log it. Returns an error if no JSON-shaped line // arrived on stdout. func (e Env) RunStats(ctx context.Context, handle LineHandler) (*RepoStats, error) { - cmd := exec.CommandContext(ctx, e.Bin, "stats", "--json", "--mode", "raw-data") - cmd.Env = e.envSlice() - cmd.Dir = e.WorkDir + cmd := e.resticCmd(ctx, "stats", "--json", "--mode", "raw-data") var out *RepoStats capture := func(stream, line string, ev any) { if stream == "stdout" && strings.HasPrefix(line, "{") { @@ -378,9 +395,7 @@ func (e Env) RunCheck(ctx context.Context, subsetPct int, handle LineHandler) (C if subsetPct > 0 { args = append(args, "--read-data-subset", fmt.Sprintf("%d%%", subsetPct)) } - cmd := exec.CommandContext(ctx, e.Bin, args...) - cmd.Env = e.envSlice() - cmd.Dir = e.WorkDir + cmd := e.resticCmd(ctx, args...) var res CheckResult sniff := func(stream, line string, ev any) { diff --git a/internal/restic/runner_test.go b/internal/restic/runner_test.go index a2d6708..698a122 100644 --- a/internal/restic/runner_test.go +++ b/internal/restic/runner_test.go @@ -174,6 +174,43 @@ func TestRunStatsErrorsWithoutJSON(t *testing.T) { } } +func TestBandwidthLimitFlagsInjected(t *testing.T) { + // Script echoes its argv to stdout. Each variant should produce + // the right --limit-* flags before the subcommand. + cases := []struct { + name string + env Env + want []string + }{ + {"both caps", Env{LimitUploadKBps: 1024, LimitDownloadKBps: 512}, []string{"--limit-upload 1024", "--limit-download 512"}}, + {"only upload", Env{LimitUploadKBps: 256}, []string{"--limit-upload 256"}}, + {"zero means omit", Env{LimitUploadKBps: 0, LimitDownloadKBps: 0}, nil}, + {"negative means omit", Env{LimitUploadKBps: -1}, nil}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + bin := setupScriptBin(t, `echo "$@"`) + env := c.env + env.Bin = bin + lines, h := captureLines() + if err := env.RunUnlock(context.Background(), h); err != nil { + t.Fatalf("RunUnlock: %v", err) + } + joined := strings.Join(*lines, "\n") + for _, want := range c.want { + if !strings.Contains(joined, want) { + t.Fatalf("want %q in argv; got: %s", want, joined) + } + } + if len(c.want) == 0 { + if strings.Contains(joined, "--limit-upload") || strings.Contains(joined, "--limit-download") { + t.Fatalf("expected no limit flags; got: %s", joined) + } + } + }) + } +} + func TestRunStatsZeroSnapshots(t *testing.T) { // Confirms RunStats succeeds and returns a valid *RepoStats when the // repo has no snapshots (snapshots_count=0). A regression that diff --git a/internal/server/http/host_bandwidth.go b/internal/server/http/host_bandwidth.go index e42996b..8165a09 100644 --- a/internal/server/http/host_bandwidth.go +++ b/internal/server/http/host_bandwidth.go @@ -58,5 +58,10 @@ func (s *Server) handleUpdateHostBandwidth(w stdhttp.ResponseWriter, r *stdhttp. writeJSONError(w, stdhttp.StatusInternalServerError, "internal", err.Error()) return } + // Fan out to the agent if connected. Errors are non-fatal — the + // next reconnect's onAgentHello will resync. + if s.deps.Hub != nil && s.deps.Hub.Connected(hostID) { + _ = s.pushBandwidthToAgent(r.Context(), hostID, req.BandwidthUpKBps, req.BandwidthDownKBps) + } writeJSON(w, stdhttp.StatusOK, hostBandwidthView(req)) } diff --git a/internal/server/http/host_bandwidth_push.go b/internal/server/http/host_bandwidth_push.go new file mode 100644 index 0000000..2cfa7fd --- /dev/null +++ b/internal/server/http/host_bandwidth_push.go @@ -0,0 +1,78 @@ +// host_bandwidth_push.go — server → agent fan-out of host-wide +// bandwidth caps via config.update. +// +// Two entry points: pushBandwidthOnHello (called from onAgentHello, +// always pushes the current state so the agent picks up edits made +// while it was offline) and pushBandwidthToAgent (called after the +// PUT bandwidth handler succeeds, so an online agent re-arms within +// seconds). +// +// We always send pointer fields (zero-valued when uncapped) so the +// agent can distinguish "no change" (nil → field absent on the wire) +// from "explicitly cleared" (non-nil zero pointer). See +// api.ConfigUpdatePayload doc for the wire semantics. +package http + +import ( + "context" + "log/slog" + "time" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/api" + "gitea.dcglab.co.uk/steve/restic-manager/internal/server/ws" +) + +// pushBandwidthOnHello ships the host's current bandwidth caps as a +// config.update on the supplied conn. Silent no-op on lookup error. +func (s *Server) pushBandwidthOnHello(ctx context.Context, hostID string, conn *ws.Conn) { + host, err := s.deps.Store.GetHost(ctx, hostID) + if err != nil { + slog.Warn("on-hello: load host for bandwidth", "host_id", hostID, "err", err) + return + } + payload := bandwidthPayload(host.BandwidthUpKBps, host.BandwidthDownKBps) + env, err := api.Marshal(api.MsgConfigUpdate, "", payload) + if err != nil { + slog.Error("on-hello: marshal bandwidth config.update", "host_id", hostID, "err", err) + return + } + sendCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + if err := conn.Send(sendCtx, env); err != nil { + slog.Warn("on-hello: send bandwidth config.update", "host_id", hostID, "err", err) + } +} + +// pushBandwidthToAgent ships the supplied caps via the hub. Caller is +// expected to check Hub.Connected first when it matters. +func (s *Server) pushBandwidthToAgent(ctx context.Context, hostID string, up, down *int) error { + env, err := api.Marshal(api.MsgConfigUpdate, "", bandwidthPayload(up, down)) + if err != nil { + return err + } + sendCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + return s.deps.Hub.Send(sendCtx, hostID, env) +} + +// bandwidthPayload builds a ConfigUpdatePayload with only the +// bandwidth fields populated. Pointers are passed through verbatim; +// callers wanting to clear a cap should pass a non-nil pointer to 0. +// On the on-hello path we materialize zero-valued pointers when the +// host record has no cap set, so the agent's stored state is always +// in sync (rather than retaining whatever value it last received). +func bandwidthPayload(up, down *int) api.ConfigUpdatePayload { + zero := 0 + upPtr := up + if upPtr == nil { + upPtr = &zero + } + downPtr := down + if downPtr == nil { + downPtr = &zero + } + return api.ConfigUpdatePayload{ + BandwidthUpKBps: upPtr, + BandwidthDownKBps: downPtr, + } +} diff --git a/internal/server/http/host_credentials.go b/internal/server/http/host_credentials.go index 0060de3..c414eba 100644 --- a/internal/server/http/host_credentials.go +++ b/internal/server/http/host_credentials.go @@ -399,6 +399,10 @@ func (s *Server) pushAdminCredsToAgent(ctx context.Context, hostID string) error // don't race a brand-new register against an old still-closing conn. func (s *Server) onAgentHello(ctx context.Context, hostID string, conn *ws.Conn) { s.pushRepoCredsOnHello(ctx, hostID, conn) + // Bandwidth caps are sent unconditionally so an agent that + // reconnects after a cap edit picks up the new state without + // waiting for the next bandwidth PUT. + s.pushBandwidthOnHello(ctx, hostID, conn) // Push the current schedule set in the same on-hello window so // the agent's local cron is in sync before any command.run lands. // An empty schedule list is a valid push: it tells the agent to From e6fc9e99634f15fcb4eb8855d7ab6146939c15e2 Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Mon, 4 May 2026 10:41:13 +0100 Subject: [PATCH 02/15] ui+server: per-job bandwidth override on Run-now P2R-13b. POST /hosts/{id}/source-groups/{gid}/run accepts optional bandwidth_up_kbps / bandwidth_down_kbps form fields, plumbs them onto CommandRunPayload. Agent dispatcher already prefers per-job override over host-wide caps (T1). UI wraps the Run-now button in a form with a
'Limit bandwidth for this run' disclosure containing two KB/s inputs. --- internal/server/http/run_group.go | 47 ++++++- .../server/http/run_group_bandwidth_test.go | 133 ++++++++++++++++++ web/templates/pages/host_sources.html | 25 +++- 3 files changed, 197 insertions(+), 8 deletions(-) create mode 100644 internal/server/http/run_group_bandwidth_test.go diff --git a/internal/server/http/run_group.go b/internal/server/http/run_group.go index 1a0f35c..2b8d952 100644 --- a/internal/server/http/run_group.go +++ b/internal/server/http/run_group.go @@ -9,6 +9,7 @@ package http import ( "errors" stdhttp "net/http" + "strconv" "github.com/go-chi/chi/v5" @@ -16,6 +17,34 @@ import ( "gitea.dcglab.co.uk/steve/restic-manager/internal/store" ) +// parseBandwidthOverride pulls optional bandwidth_up_kbps / +// bandwidth_down_kbps from the request (form or query). Returns nil +// for any field absent or empty; an explicit "0" produces a non-nil +// pointer to 0 — i.e., "no cap for this run, even if the host has +// one set." Non-integers / negatives are rejected with an error. +func parseBandwidthOverride(r *stdhttp.Request) (up *int, down *int, err error) { + parse := func(name string) (*int, error) { + v := r.FormValue(name) + if v == "" { + return nil, nil + } + n, perr := strconv.Atoi(v) + if perr != nil { + return nil, errors.New(name + " must be an integer") + } + if n < 0 { + return nil, errors.New(name + " must be >= 0") + } + return &n, nil + } + up, err = parse("bandwidth_up_kbps") + if err != nil { + return nil, nil, err + } + down, err = parse("bandwidth_down_kbps") + return up, down, err +} + func (s *Server) handleRunSourceGroup(w stdhttp.ResponseWriter, r *stdhttp.Request) { user, ok := s.requireUser(r) if !ok { @@ -40,13 +69,25 @@ func (s *Server) handleRunSourceGroup(w stdhttp.ResponseWriter, r *stdhttp.Reque return } + // Optional per-run bandwidth override. Disclosed in the UI under a + //
"Limit bandwidth for this run" affordance; absent on + // the wire (and from JSON callers that don't supply it) means + // "fall back to the host's standing caps." + upOverride, downOverride, perr := parseBandwidthOverride(r) + if perr != nil { + s.runGroupError(w, r, stdhttp.StatusBadRequest, "invalid_value", perr.Error()) + return + } + // Backup invocations don't consume RetentionPolicy — that lives on // forget. Sending the resolved set here would just be dead weight. res, status, code, msg := s.dispatchJobWithPayload(r.Context(), user, hostID, api.JobBackup, api.CommandRunPayload{ - Includes: g.Includes, - Excludes: g.Excludes, - Tag: g.Name, + Includes: g.Includes, + Excludes: g.Excludes, + Tag: g.Name, + BandwidthUpKBps: upOverride, + BandwidthDownKBps: downOverride, }) if code != "" { s.runGroupError(w, r, status, code, msg) diff --git a/internal/server/http/run_group_bandwidth_test.go b/internal/server/http/run_group_bandwidth_test.go new file mode 100644 index 0000000..36c01a3 --- /dev/null +++ b/internal/server/http/run_group_bandwidth_test.go @@ -0,0 +1,133 @@ +// run_group_bandwidth_test.go — covers the per-job bandwidth override +// that operators can set via the Run-now form's "Limit bandwidth for +// this run" disclosure (P2R-13b). +package http + +import ( + "context" + "encoding/json" + stdhttp "net/http" + "net/url" + "strings" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/oklog/ulid/v2" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/api" + "gitea.dcglab.co.uk/steve/restic-manager/internal/store" +) + +// TestRunSourceGroupBandwidthOverride: connect a fake agent, POST the +// per-group Run-now endpoint with bandwidth_up_kbps=512, assert the +// dispatched command.run carries it. +func TestRunSourceGroupBandwidthOverride(t *testing.T) { + t.Parallel() + srv, ts, st := rawTestServer(t) + hostID, token := enrolHostForWS(t, srv, st, "bw-host") + + // Pre-seed an init job so auto-init doesn't fire on hello and + // pollute our envelope sequence. + if err := st.CreateJob(context.Background(), store.Job{ + ID: ulid.Make().String(), HostID: hostID, Kind: "init", + ActorKind: "system", CreatedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("seed init: %v", err) + } + + gid := ulid.Make().String() + if err := st.CreateSourceGroup(context.Background(), &store.SourceGroup{ + ID: gid, HostID: hostID, Name: "etc", Includes: []string{"/etc"}, + }); err != nil { + t.Fatalf("group: %v", err) + } + + c := agentDial(t, srv, ts, hostID, token) + sendHello(t, c, "bw-host") + // Drain on-hello burst before issuing the run-now. + _ = drainUntil(t, c, api.MsgScheduleSet) + + cookie := loginAsAdmin(t, st) + form := url.Values{ + "bandwidth_up_kbps": {"512"}, + "bandwidth_down_kbps": {"256"}, + } + req, _ := stdhttp.NewRequest("POST", + ts.URL+"/hosts/"+hostID+"/source-groups/"+gid+"/run", + strings.NewReader(form.Encode())) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.Header.Set("Accept", "application/json") + req.AddCookie(cookie) + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + res.Body.Close() + if res.StatusCode != stdhttp.StatusAccepted { + t.Fatalf("status: got %d, want 202", res.StatusCode) + } + + // Read the dispatched command.run; assert overrides are present. + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + ctx, cancel := context.WithTimeout(context.Background(), 800*time.Millisecond) + mt, raw, rerr := c.Read(ctx) + cancel() + if rerr != nil { + break + } + if mt != websocket.MessageText { + continue + } + var env api.Envelope + _ = json.Unmarshal(raw, &env) + if env.Type != api.MsgCommandRun { + continue + } + var p api.CommandRunPayload + if err := env.UnmarshalPayload(&p); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if p.Kind != api.JobBackup { + continue + } + if p.BandwidthUpKBps == nil || *p.BandwidthUpKBps != 512 { + t.Fatalf("BandwidthUpKBps: got %v, want 512", p.BandwidthUpKBps) + } + if p.BandwidthDownKBps == nil || *p.BandwidthDownKBps != 256 { + t.Fatalf("BandwidthDownKBps: got %v, want 256", p.BandwidthDownKBps) + } + return + } + t.Fatal("timed out waiting for command.run with bandwidth override") +} + +// TestRunSourceGroupBandwidthRejectsNegative: invalid value → 400. +func TestRunSourceGroupBandwidthRejectsNegative(t *testing.T) { + t.Parallel() + _, url2, st := newTestServerWithHub(t) + cookie := loginAsAdmin(t, st) + hostID := makeHost(t, st, "bw-rej-host") + gid := ulid.Make().String() + if err := st.CreateSourceGroup(context.Background(), &store.SourceGroup{ + ID: gid, HostID: hostID, Name: "etc", Includes: []string{"/etc"}, + }); err != nil { + t.Fatalf("group: %v", err) + } + form := url.Values{"bandwidth_up_kbps": {"-1"}} + req, _ := stdhttp.NewRequest("POST", + url2+"/hosts/"+hostID+"/source-groups/"+gid+"/run", + strings.NewReader(form.Encode())) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.Header.Set("Accept", "application/json") + req.AddCookie(cookie) + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + defer res.Body.Close() + if res.StatusCode != stdhttp.StatusBadRequest { + t.Fatalf("status: got %d, want 400", res.StatusCode) + } +} diff --git a/web/templates/pages/host_sources.html b/web/templates/pages/host_sources.html index 36a8077..d0d1087 100644 --- a/web/templates/pages/host_sources.html +++ b/web/templates/pages/host_sources.html @@ -53,12 +53,27 @@ {{if gt $row.SnapshotCount 0}} · {{$row.SnapshotCount}} snapshot{{if ne $row.SnapshotCount 1}}s{{end}}{{end}} -
+
{{if and (gt (len $g.Includes) 0) (eq $host.Status "online")}} - +
+ +
+ Limit bandwidth for this run +
+ + + + + KB/s +
+
+
{{else}} From d02a093eeb1a75c76de2e587f869320350510ecc Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Mon, 4 May 2026 10:44:31 +0100 Subject: [PATCH 03/15] ui+server: schedule next-run / last-run on dashboard + schedules tab P2R-14. New store.LatestJobBySchedule query (per-schedule fired job). Schedules-tab handler computes next-fire from cron + last-fire from the jobs table per row. Schedules table grows two columns; dashboard host row prepends 'next 12h ago/from now' to the existing last-backup line when a single covering schedule is the run-now candidate. Embeds store.Schedule into scheduleRow so existing template field references keep working without bulk renames. --- internal/server/http/schedule_nextrun_test.go | 48 ++++++++++ internal/server/http/ui_handlers.go | 14 +++ internal/server/http/ui_schedules.go | 38 +++++++- internal/store/schedule_runs.go | 88 +++++++++++++++++++ web/styles/input.css | 4 +- web/templates/pages/host_schedules.html | 10 +++ web/templates/partials/host_row.html | 3 + 7 files changed, 201 insertions(+), 4 deletions(-) create mode 100644 internal/server/http/schedule_nextrun_test.go create mode 100644 internal/store/schedule_runs.go diff --git a/internal/server/http/schedule_nextrun_test.go b/internal/server/http/schedule_nextrun_test.go new file mode 100644 index 0000000..6088f1f --- /dev/null +++ b/internal/server/http/schedule_nextrun_test.go @@ -0,0 +1,48 @@ +// schedule_nextrun_test.go — pin the cron parser → next-run shape we +// rely on for the dashboard host row + schedules tab (P2R-14). +package http + +import ( + "testing" + "time" +) + +func TestCronParserNext(t *testing.T) { + cases := []struct { + name string + expr string + from time.Time + want time.Time + }{ + { + name: "daily at 03:00", + expr: "0 3 * * *", + from: time.Date(2026, 5, 4, 1, 0, 0, 0, time.UTC), + want: time.Date(2026, 5, 4, 3, 0, 0, 0, time.UTC), + }, + { + name: "daily at 03:00 (after time of day → next day)", + expr: "0 3 * * *", + from: time.Date(2026, 5, 4, 5, 0, 0, 0, time.UTC), + want: time.Date(2026, 5, 5, 3, 0, 0, 0, time.UTC), + }, + { + name: "every 15 minutes", + expr: "*/15 * * * *", + from: time.Date(2026, 5, 4, 1, 7, 0, 0, time.UTC), + want: time.Date(2026, 5, 4, 1, 15, 0, 0, time.UTC), + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + parsed, err := cronParser.Parse(c.expr) + if err != nil { + t.Fatalf("parse %q: %v", c.expr, err) + } + got := parsed.Next(c.from) + if !got.Equal(c.want) { + t.Fatalf("Next(%v) = %v, want %v", c.from, got, c.want) + } + }) + } +} diff --git a/internal/server/http/ui_handlers.go b/internal/server/http/ui_handlers.go index 0dd4752..cd4166d 100644 --- a/internal/server/http/ui_handlers.go +++ b/internal/server/http/ui_handlers.go @@ -124,6 +124,9 @@ type dashboardHostRow struct { // match — in that case the row shows "Open →" instead of a Run-now // button (the operator picks per-group from the host detail). RunAllScheduleID string + // NextRun is the next-fire time of RunAllScheduleID (when set), + // computed server-side from its cron. nil otherwise. + NextRun *time.Time } // pickRunAllSchedule returns the ID of the single schedule whose @@ -203,6 +206,17 @@ func (s *Server) handleUIDashboard(w stdhttp.ResponseWriter, r *stdhttp.Request) slog.Warn("ui dashboard: list schedules", "host_id", h.ID, "err", serr) } row.RunAllScheduleID = pickRunAllSchedule(scheds, groups) + if row.RunAllScheduleID != "" { + for _, sc := range scheds { + if sc.ID == row.RunAllScheduleID { + if parsed, perr := cronParser.Parse(sc.CronExpr); perr == nil { + n := parsed.Next(time.Now().UTC()).UTC() + row.NextRun = &n + } + break + } + } + } rows = append(rows, row) } diff --git a/internal/server/http/ui_schedules.go b/internal/server/http/ui_schedules.go index 485e64e..a4daf4d 100644 --- a/internal/server/http/ui_schedules.go +++ b/internal/server/http/ui_schedules.go @@ -25,10 +25,22 @@ import ( // the template doesn't need to do per-row store lookups. type hostSchedulesPage struct { hostChromeData - Schedules []store.Schedule + Schedules []scheduleRow GroupNames map[string]string } +// scheduleRow bundles a schedule with its derived "next run" + "last +// run" data. The Schedule is embedded so existing template field +// references (`$sc.ID`, `$sc.CronExpr`, etc) keep working when we +// switch the iterating slice from []store.Schedule to []scheduleRow. +type scheduleRow struct { + store.Schedule + NextRun *time.Time + LastRun *time.Time + LastJobID string + LastStatus string // succeeded|failed|running|queued — empty when never fired +} + // scheduleFormData mirrors the form's wire shape — strings + bool for // round-trip on validation re-render. type scheduleFormData struct { @@ -74,6 +86,28 @@ func (s *Server) handleUISchedulesList(w stdhttp.ResponseWriter, r *stdhttp.Requ names[g.ID] = g.Name } + now := time.Now().UTC() + rows := make([]scheduleRow, 0, len(scheds)) + for _, sc := range scheds { + row := scheduleRow{Schedule: sc} + if sc.Enabled { + if sched, err := cronParser.Parse(sc.CronExpr); err == nil { + next := sched.Next(now).UTC() + row.NextRun = &next + } + } + if j, jerr := s.deps.Store.LatestJobBySchedule(r.Context(), host.ID, sc.ID); jerr == nil && j != nil { + t := j.CreatedAt + if j.StartedAt != nil { + t = *j.StartedAt + } + row.LastRun = &t + row.LastJobID = j.ID + row.LastStatus = j.Status + } + rows = append(rows, row) + } + chrome := s.loadHostChrome(r, *host, "schedules", "schedules") chrome.ScheduleCount = len(scheds) chrome.SourceGroupCount = len(groups) @@ -82,7 +116,7 @@ func (s *Server) handleUISchedulesList(w stdhttp.ResponseWriter, r *stdhttp.Requ view.Title = host.Name + " schedules · restic-manager" view.Page = hostSchedulesPage{ hostChromeData: chrome, - Schedules: scheds, + Schedules: rows, GroupNames: names, } if err := s.deps.UI.Render(w, "host_schedules", view); err != nil { diff --git a/internal/store/schedule_runs.go b/internal/store/schedule_runs.go new file mode 100644 index 0000000..a42f504 --- /dev/null +++ b/internal/store/schedule_runs.go @@ -0,0 +1,88 @@ +// schedule_runs.go — derived "next run" / "last run" helpers for the +// dashboard host row + schedules tab (P2R-14). +// +// Both are derived data: NextRun is computed from the cron expression +// at request time; LatestJobBySchedule reads the most recent job that +// fired against this schedule. Neither is persisted — the cost of the +// query is small relative to a page render. +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" +) + +// LatestJobBySchedule returns the most recent job fired by this +// schedule (actor_kind='schedule' AND scheduled_id=schedID), or +// (nil, ErrNotFound) when the schedule has never fired. Includes +// queued/running rows because the operator wants to see "running +// now" too. +func (s *Store) LatestJobBySchedule(ctx context.Context, hostID, schedID string) (*Job, error) { + row := s.db.QueryRowContext(ctx, + `SELECT id, host_id, kind, status, scheduled_id, actor_kind, actor_id, + started_at, finished_at, exit_code, stats, error, created_at + FROM jobs + WHERE host_id = ? AND scheduled_id = ? AND actor_kind = 'schedule' + ORDER BY created_at DESC + LIMIT 1`, hostID, schedID) + return scanJobRow(row) +} + +// scanJobRow is the shared scan used by LatestJobBySchedule. Mirrors +// the columns LatestJobByKind reads. Kept in this file (vs jobs.go) +// to avoid disturbing the stable API surface exported there. +func scanJobRow(row *sql.Row) (*Job, error) { + var ( + j Job + schedID sql.NullString + actorID sql.NullString + startedAt sql.NullString + finishedAt sql.NullString + exitCode sql.NullInt64 + stats sql.NullString + errMsg sql.NullString + createdAt string + ) + if err := row.Scan(&j.ID, &j.HostID, &j.Kind, &j.Status, &schedID, + &j.ActorKind, &actorID, &startedAt, &finishedAt, + &exitCode, &stats, &errMsg, &createdAt); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrNotFound + } + return nil, fmt.Errorf("store: scan job: %w", err) + } + if schedID.Valid { + v := schedID.String + j.ScheduledID = &v + } + if actorID.Valid { + v := actorID.String + j.ActorID = &v + } + if startedAt.Valid { + t, _ := time.Parse(time.RFC3339Nano, startedAt.String) + j.StartedAt = &t + } + if finishedAt.Valid { + t, _ := time.Parse(time.RFC3339Nano, finishedAt.String) + j.FinishedAt = &t + } + if exitCode.Valid { + i := int(exitCode.Int64) + j.ExitCode = &i + } + if stats.Valid && stats.String != "" { + j.Stats = []byte(stats.String) + } + if errMsg.Valid { + v := errMsg.String + j.Error = &v + } + if t, err := time.Parse(time.RFC3339Nano, createdAt); err == nil { + j.CreatedAt = t + } + return &j, nil +} diff --git a/web/styles/input.css b/web/styles/input.css index 565c5e2..7f0ff24 100644 --- a/web/styles/input.css +++ b/web/styles/input.css @@ -209,8 +209,8 @@ /* ---------- schedule rows (Schedules tab) ---------- */ .schd-row { display: grid; align-items: center; - grid-template-columns: 90px 1fr 2fr auto; - column-gap: 18px; + grid-template-columns: 78px 1fr 1.6fr 100px 110px auto; + column-gap: 14px; padding: 12px 18px; font-size: 13px; } .schd-row.head { diff --git a/web/templates/pages/host_schedules.html b/web/templates/pages/host_schedules.html index 764ae2d..4a49e79 100644 --- a/web/templates/pages/host_schedules.html +++ b/web/templates/pages/host_schedules.html @@ -33,6 +33,8 @@
Status
Cron
Sources
+
Next
+
Last
{{range $i, $sc := $page.Schedules}} @@ -52,6 +54,14 @@ {{if $name}}{{$name}}{{else}}unknown{{end}} {{end}}
+
+ {{if $sc.NextRun}}{{relTime $sc.NextRun}}{{else if not $sc.Enabled}}(paused){{else}}—{{end}} +
+
+ {{if $sc.LastRun}}{{relTime $sc.LastRun}}{{else}}—{{end}} +
{{if eq $host.Status "online"}} {{if $sc.Enabled}} diff --git a/web/templates/partials/host_row.html b/web/templates/partials/host_row.html index 98b27ea..9c7799e 100644 --- a/web/templates/partials/host_row.html +++ b/web/templates/partials/host_row.html @@ -30,6 +30,9 @@ {{- else -}} never run {{- end -}} + {{- if .NextRun -}} +
next {{relTime .NextRun}} + {{- end -}}
{{bytes $h.RepoSizeBytes}}
From c9b49637d1220b385bb9d0c4b83e2baa74bf8483 Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Mon, 4 May 2026 10:49:57 +0100 Subject: [PATCH 04/15] =?UTF-8?q?ui:=20P2R-09=20auto-init=20UX=20=E2=80=94?= =?UTF-8?q?=20init=20line=20in=20chrome=20+=20danger-zone=20re-init?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Latest 'init' job status surfaced under the host-detail vitals strip (succeeded/failed/running/queued, with link to the live job log on non-success). New POST /hosts/{id}/repo/reinit handler dispatches a fresh init job after the operator types the host name to confirm; audit row records 'host.repo_reinit'. --- internal/server/http/server.go | 1 + internal/server/http/ui_handlers.go | 15 ++ internal/server/http/ui_repo_reinit.go | 120 +++++++++++ internal/server/http/ui_repo_reinit_test.go | 212 ++++++++++++++++++++ web/templates/pages/host_repo.html | 12 +- web/templates/partials/host_chrome.html | 16 ++ 6 files changed, 374 insertions(+), 2 deletions(-) create mode 100644 internal/server/http/ui_repo_reinit.go create mode 100644 internal/server/http/ui_repo_reinit_test.go diff --git a/internal/server/http/server.go b/internal/server/http/server.go index 8ef3d83..02adfad 100644 --- a/internal/server/http/server.go +++ b/internal/server/http/server.go @@ -228,6 +228,7 @@ func (s *Server) routes(r chi.Router) { r.Post("/hosts/{id}/repo/credentials", s.handleUIRepoCredentialsSave) r.Post("/hosts/{id}/repo/bandwidth", s.handleUIRepoBandwidthSave) r.Post("/hosts/{id}/repo/maintenance", s.handleUIRepoMaintenanceSave) + r.Post("/hosts/{id}/repo/reinit", s.handleUIRepoReinit) // Admin credentials form (separate slot for prune-capable user). r.Post("/hosts/{id}/admin-credentials", s.handleUIAdminCredentialsSave) r.Post("/hosts/{id}/admin-credentials/delete", s.handleUIAdminCredentialsDelete) diff --git a/internal/server/http/ui_handlers.go b/internal/server/http/ui_handlers.go index cd4166d..ad23e03 100644 --- a/internal/server/http/ui_handlers.go +++ b/internal/server/http/ui_handlers.go @@ -499,6 +499,12 @@ type hostChromeData struct { SourceGroupCount int ScheduleCount int ScheduleVersion int64 // host_schedule_version (latest desired) + + // Auto-init status surfaced from the latest 'init' job. + // InitStatus is "succeeded" | "failed" | "running" | "queued" | "" (never run). + InitStatus string + InitAt *time.Time // started_at if non-nil else created_at + InitJobID string } // loadHostChrome fetches the per-tab counts that every host-detail tab @@ -520,6 +526,15 @@ func (s *Server) loadHostChrome(r *stdhttp.Request, host store.Host, subtab, cru if v, err := s.deps.Store.GetHostScheduleVersion(r.Context(), host.ID); err == nil { d.ScheduleVersion = v } + if j, err := s.deps.Store.LatestJobByKind(r.Context(), host.ID, "init"); err == nil && j != nil { + d.InitStatus = j.Status + d.InitJobID = j.ID + t := j.CreatedAt + if j.StartedAt != nil { + t = *j.StartedAt + } + d.InitAt = &t + } return d } diff --git a/internal/server/http/ui_repo_reinit.go b/internal/server/http/ui_repo_reinit.go new file mode 100644 index 0000000..c817df0 --- /dev/null +++ b/internal/server/http/ui_repo_reinit.go @@ -0,0 +1,120 @@ +// ui_repo_reinit.go — danger-zone re-init handler. Dispatches a fresh +// `restic init` job after the operator types the host name to confirm. +// Restic itself refuses to overwrite an existing repo (its init is +// effectively idempotent — see the runner's "config file already +// exists" sniff in restic.RunInit), so this is *not* a destructive +// data wipe; it's a "try again from scratch" affordance for the +// operator. If the rest-server bucket needs clearing the operator +// has to do that out-of-band; the job log will say so. +// +// Audit-logged with action='host.repo_reinit' so the trail records +// who triggered the wipe attempt and when. +package http + +import ( + "context" + "errors" + "log/slog" + stdhttp "net/http" + "strings" + "time" + + "github.com/oklog/ulid/v2" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/api" + "gitea.dcglab.co.uk/steve/restic-manager/internal/store" +) + +func (s *Server) handleUIRepoReinit(w stdhttp.ResponseWriter, r *stdhttp.Request) { + u := s.requireUIUser(w, r) + if u == nil { + return + } + host, ok := s.loadHostForUI(w, r) + if !ok { + return + } + if err := r.ParseForm(); err != nil { + stdhttp.Error(w, "bad request", stdhttp.StatusBadRequest) + return + } + confirm := strings.TrimSpace(r.PostForm.Get("confirm_hostname")) + if confirm != host.Name { + // We don't have a dedicated re-init banner field; surface via + // the existing CredentialsError slot — it sits adjacent to the + // danger zone visually so the operator's eye lands on it. + s.renderRepoPage(w, r, u, host, + "Re-init aborted — typed hostname did not match.", "", "", "") + return + } + if !s.deps.Hub.Connected(host.ID) { + s.renderRepoPage(w, r, u, host, + "Host is offline — bring the agent back up before re-initializing.", + "", "", "") + return + } + // Ensure the host has creds bound; otherwise restic init can't + // connect to the repo. + if _, err := s.deps.Store.GetHostCredentials(r.Context(), host.ID, store.CredKindRepo); err != nil { + if errors.Is(err, store.ErrNotFound) { + s.renderRepoPage(w, r, u, host, + "Bind repo credentials before re-initializing.", + "", "", "") + return + } + stdhttp.Error(w, "internal", stdhttp.StatusInternalServerError) + return + } + + jobID := ulid.Make().String() + now := time.Now().UTC() + if err := s.deps.Store.CreateJob(r.Context(), store.Job{ + ID: jobID, + HostID: host.ID, + Kind: string(api.JobInit), + ActorKind: "user", + ActorID: &u.ID, + CreatedAt: now, + }); err != nil { + slog.Error("repo reinit: persist job", "host_id", host.ID, "err", err) + stdhttp.Error(w, "internal", stdhttp.StatusInternalServerError) + return + } + + env, err := api.Marshal(api.MsgCommandRun, jobID, api.CommandRunPayload{ + JobID: jobID, + Kind: api.JobInit, + }) + if err != nil { + stdhttp.Error(w, "internal", stdhttp.StatusInternalServerError) + return + } + sendCtx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + if err := s.deps.Hub.Send(sendCtx, host.ID, env); err != nil { + slog.Warn("repo reinit: ws send failed", "host_id", host.ID, "err", err) + s.renderRepoPage(w, r, u, host, + "Failed to deliver the init job to the agent — try again.", + "", "", "") + return + } + + uid := u.ID + _ = s.deps.Store.AppendAudit(r.Context(), store.AuditEntry{ + ID: ulid.Make().String(), + UserID: &uid, + Actor: "user", + Action: "host.repo_reinit", + TargetKind: ptr("host"), + TargetID: &host.ID, + TS: now, + }) + + // HTMX redirect → live job log. JSON callers get a 202. + if wantsHTML(r) { + w.Header().Set("HX-Redirect", "/jobs/"+jobID) + w.WriteHeader(stdhttp.StatusNoContent) + return + } + stdhttp.Redirect(w, r, "/jobs/"+jobID, stdhttp.StatusSeeOther) +} diff --git a/internal/server/http/ui_repo_reinit_test.go b/internal/server/http/ui_repo_reinit_test.go new file mode 100644 index 0000000..5b893ec --- /dev/null +++ b/internal/server/http/ui_repo_reinit_test.go @@ -0,0 +1,212 @@ +// ui_repo_reinit_test.go — covers the danger-zone re-init handler: +// hostname-confirm gate + offline guard + missing-creds guard. +package http + +import ( + "context" + stdhttp "net/http" + "net/http/httptest" + "net/url" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/oklog/ulid/v2" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/api" + "gitea.dcglab.co.uk/steve/restic-manager/internal/auth" + "gitea.dcglab.co.uk/steve/restic-manager/internal/crypto" + "gitea.dcglab.co.uk/steve/restic-manager/internal/server/config" + "gitea.dcglab.co.uk/steve/restic-manager/internal/server/ui" + "gitea.dcglab.co.uk/steve/restic-manager/internal/server/ws" + "gitea.dcglab.co.uk/steve/restic-manager/internal/store" +) + +// rawTestServerWithUI is the rawTestServer twin that also wires the +// UI renderer in, returning the raw httptest server so callers can +// dial /ws/agent. The UI is needed for the repo-reinit handler's +// error re-render path. +func rawTestServerWithUI(t *testing.T) (*Server, *httptest.Server, *store.Store) { + t.Helper() + dir := t.TempDir() + st, err := store.Open(context.Background(), filepath.Join(dir, "rm.db")) + if err != nil { + t.Fatalf("store: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + keyPath := filepath.Join(dir, "secret.key") + _ = crypto.GenerateKeyFile(keyPath) + key, _ := crypto.LoadKeyFromFile(keyPath) + aead, _ := crypto.NewAEAD(key) + renderer, err := ui.New() + if err != nil { + t.Fatalf("ui.New: %v", err) + } + deps := Deps{ + Cfg: config.Config{Listen: ":0", DataDir: dir, SecretKeyFile: keyPath}, + Store: st, + AEAD: aead, + Hub: ws.NewHub(), + UI: renderer, + } + srv := New(deps) + ts := httptest.NewServer(srv.srv.Handler) + t.Cleanup(ts.Close) + return srv, ts, st +} + +// enrolHostForUI is the enrolHostForWS twin for tests that use the +// UI-enabled rawTestServerWithUI. +func enrolHostForUI(t *testing.T, _ *Server, st *store.Store, name string) (hostID, token string) { + t.Helper() + hostID = ulid.Make().String() + token, _ = auth.NewToken() + if err := st.CreateHost(context.Background(), store.Host{ + ID: hostID, Name: name, OS: "linux", Arch: "amd64", + EnrolledAt: time.Now().UTC(), + }, auth.HashToken(token), ""); err != nil { + t.Fatalf("create host: %v", err) + } + return hostID, token +} + +// TestRepoReinitWrongHostnameRejected: typing a different name keeps +// the page on the repo screen with an error banner; no init job is +// dispatched. +func TestRepoReinitWrongHostnameRejected(t *testing.T) { + t.Parallel() + srv, ts, st := rawTestServerWithUI(t) + hostID, token := enrolHostForUI(t, srv, st, "reinit-host") + + c := agentDial(t, srv, ts, hostID, token) + sendHello(t, c, "reinit-host") + _ = drainUntil(t, c, api.MsgScheduleSet) + + cookie := loginAsAdmin(t, st) + + form := url.Values{"confirm_hostname": {"WRONG-NAME"}} + req, _ := stdhttp.NewRequest("POST", + ts.URL+"/hosts/"+hostID+"/repo/reinit", + strings.NewReader(form.Encode())) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.AddCookie(cookie) + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + defer res.Body.Close() + if res.StatusCode != stdhttp.StatusUnprocessableEntity { + t.Fatalf("status: got %d, want 422 (re-rendered page with banner)", res.StatusCode) + } + // No init job should appear in the queue beyond the one auto-init + // pushed on hello (which fires when no init has run yet — let's + // just make sure no new "user" actor init was created). + var n int + if err := st.DB().QueryRow( + `SELECT COUNT(*) FROM jobs WHERE host_id = ? AND kind = 'init' AND actor_kind = 'user'`, + hostID).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + if n != 0 { + t.Fatalf("user-actor init jobs: got %d, want 0 (gate was bypassed)", n) + } +} + +// TestRepoReinitDispatchesOnMatch: typing the right hostname dispatches +// a new init job + audit row. +func TestRepoReinitDispatchesOnMatch(t *testing.T) { + t.Parallel() + srv, ts, st := rawTestServerWithUI(t) + hostID, token := enrolHostForUI(t, srv, st, "reinit-ok-host") + // Bind repo creds — re-init guard requires them. + enc, err := srv.encryptRepoCreds(repoCredsBlob{ + RepoURL: "rest:http://r/x", RepoUsername: "u", RepoPassword: "p", + }, []byte("host:"+hostID)) + if err != nil { + t.Fatalf("encrypt: %v", err) + } + if err := st.SetHostCredentials(context.Background(), hostID, store.CredKindRepo, enc); err != nil { + t.Fatalf("set creds: %v", err) + } + + // Pre-seed a successful init so auto-init doesn't fire on hello. + preID := ulid.Make().String() + if err := st.CreateJob(context.Background(), store.Job{ + ID: preID, HostID: hostID, Kind: "init", + ActorKind: "system", CreatedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("seed init: %v", err) + } + if err := st.MarkJobFinished(context.Background(), preID, "succeeded", 0, nil, "", time.Now().UTC()); err != nil { + t.Fatalf("mark seed init: %v", err) + } + + c := agentDial(t, srv, ts, hostID, token) + sendHello(t, c, "reinit-ok-host") + _ = drainUntil(t, c, api.MsgScheduleSet) + + cookie := loginAsAdmin(t, st) + + form := url.Values{"confirm_hostname": {"reinit-ok-host"}} + req, _ := stdhttp.NewRequest("POST", + ts.URL+"/hosts/"+hostID+"/repo/reinit", + strings.NewReader(form.Encode())) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.Header.Set("HX-Request", "true") // get HX-Redirect path + req.AddCookie(cookie) + res, err := stdhttp.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + defer res.Body.Close() + if res.StatusCode != stdhttp.StatusNoContent { + t.Fatalf("status: got %d, want 204", res.StatusCode) + } + if res.Header.Get("HX-Redirect") == "" { + t.Fatal("expected HX-Redirect header") + } + + // Read the dispatched command.run; assert it's an init job. + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + mt, raw, rerr := c.Read(ctx) + cancel() + if rerr != nil { + break + } + if mt != websocket.MessageText { + continue + } + // Quick parse — we only care about the type. Avoid full + // envelope unmarshal here because the surrounding loop is just + // looking for the command.run we triggered. + if !strings.Contains(string(raw), `"command.run"`) { + continue + } + // Verify a user-actor init job row was created. + var n int + if err := st.DB().QueryRow( + `SELECT COUNT(*) FROM jobs WHERE host_id = ? AND kind = 'init' AND actor_kind = 'user'`, + hostID).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + if n != 1 { + t.Fatalf("user-actor init jobs: got %d, want 1", n) + } + // Audit row. + var na int + if err := st.DB().QueryRow( + `SELECT COUNT(*) FROM audit_log WHERE action = 'host.repo_reinit' AND target_id = ?`, + hostID).Scan(&na); err != nil { + t.Fatalf("audit count: %v", err) + } + if na != 1 { + t.Fatalf("audit rows: got %d, want 1", na) + } + return + } + t.Fatal("timed out waiting for command.run after re-init dispatch") +} diff --git a/web/templates/pages/host_repo.html b/web/templates/pages/host_repo.html index a0caf6e..fda3489 100644 --- a/web/templates/pages/host_repo.html +++ b/web/templates/pages/host_repo.html @@ -238,8 +238,16 @@ secrets.enc is reused.

- +
+ + +
diff --git a/web/templates/partials/host_chrome.html b/web/templates/partials/host_chrome.html index 01606de..9e3f741 100644 --- a/web/templates/partials/host_chrome.html +++ b/web/templates/partials/host_chrome.html @@ -105,6 +105,22 @@ + {{/* ---------- repo init line (P2R-09) ---------- */}} + {{if $page.InitStatus}} +
+ {{if eq $page.InitStatus "succeeded"}} + repo ready · initialised {{relTime $page.InitAt}} + {{else if eq $page.InitStatus "failed"}} + init failed · + job {{$page.InitJobID}} · retry from the Repo tab's danger zone + {{else if eq $page.InitStatus "running"}} + init running… · live log → + {{else if eq $page.InitStatus "queued"}} + init queued · job {{$page.InitJobID}} + {{end}} +
+ {{end}} + {{/* ---------- secondary tabs ---------- */}}
Snapshots {{comma $host.SnapshotCount}} From 18b0bf976dbd6f5a340bbaa32acedb92da396bc1 Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Mon, 4 May 2026 10:52:16 +0100 Subject: [PATCH 05/15] store: P2R-10 schema for source-group + host-default hooks (migration 0010) Adds pre_hook/post_hook BLOB columns to source_groups and pre_hook_default/post_hook_default to hosts. Bytes stored verbatim (AEAD encrypt/decrypt happens at the HTTP layer where the AEAD key lives). Round-trip tests cover set/clear semantics on both tables. --- internal/store/hooks_test.go | 106 +++++++++++++++++++++++ internal/store/hosts.go | 28 +++++- internal/store/migrations/0010_hooks.sql | 25 ++++++ internal/store/sources.go | 29 +++++-- internal/store/types.go | 13 +++ 5 files changed, 190 insertions(+), 11 deletions(-) create mode 100644 internal/store/hooks_test.go create mode 100644 internal/store/migrations/0010_hooks.sql diff --git a/internal/store/hooks_test.go b/internal/store/hooks_test.go new file mode 100644 index 0000000..18a7864 --- /dev/null +++ b/internal/store/hooks_test.go @@ -0,0 +1,106 @@ +// hooks_test.go — covers the pre/post hook columns added in +// migration 0010 (P2R-10): set + reload roundtrip on both +// source_groups and hosts; nil clears the column. +package store + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/oklog/ulid/v2" +) + +func newTestStore(t *testing.T) *Store { + t.Helper() + dir := t.TempDir() + st, err := Open(context.Background(), filepath.Join(dir, "rm.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + return st +} + +func makeHostInStore(t *testing.T, st *Store, name string) string { + t.Helper() + id := ulid.Make().String() + if err := st.CreateHost(context.Background(), Host{ + ID: id, Name: name, OS: "linux", Arch: "amd64", + EnrolledAt: time.Now().UTC(), + }, "tokenhash-"+id, ""); err != nil { + t.Fatalf("create host: %v", err) + } + return id +} + +func TestSourceGroupHooksRoundTrip(t *testing.T) { + t.Parallel() + st := newTestStore(t) + hostID := makeHostInStore(t, st, "hooks-host") + + g := &SourceGroup{ + ID: ulid.Make().String(), HostID: hostID, Name: "etc", + PreHook: []byte("ENC-PRE"), + PostHook: []byte("ENC-POST"), + } + if err := st.CreateSourceGroup(context.Background(), g); err != nil { + t.Fatalf("create: %v", err) + } + got, err := st.GetSourceGroup(context.Background(), hostID, g.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if string(got.PreHook) != "ENC-PRE" { + t.Fatalf("PreHook: got %q, want ENC-PRE", got.PreHook) + } + if string(got.PostHook) != "ENC-POST" { + t.Fatalf("PostHook: got %q, want ENC-POST", got.PostHook) + } + + // Update: clear PreHook, change PostHook. + got.PreHook = nil + got.PostHook = []byte("ENC-POST-2") + if err := st.UpdateSourceGroup(context.Background(), got); err != nil { + t.Fatalf("update: %v", err) + } + got, err = st.GetSourceGroup(context.Background(), hostID, g.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if got.PreHook != nil { + t.Fatalf("PreHook: want nil after clear, got %q", got.PreHook) + } + if string(got.PostHook) != "ENC-POST-2" { + t.Fatalf("PostHook: got %q, want ENC-POST-2", got.PostHook) + } +} + +func TestHostHookDefaultsRoundTrip(t *testing.T) { + t.Parallel() + st := newTestStore(t) + hostID := makeHostInStore(t, st, "host-hooks-host") + + if err := st.SetHostHooks(context.Background(), hostID, []byte("PRE"), []byte("POST")); err != nil { + t.Fatalf("set: %v", err) + } + h, err := st.GetHost(context.Background(), hostID) + if err != nil { + t.Fatalf("get: %v", err) + } + if string(h.PreHookDefault) != "PRE" || string(h.PostHookDefault) != "POST" { + t.Fatalf("after set: pre=%q post=%q", h.PreHookDefault, h.PostHookDefault) + } + // Clear by passing nil. + if err := st.SetHostHooks(context.Background(), hostID, nil, nil); err != nil { + t.Fatalf("clear: %v", err) + } + h, err = st.GetHost(context.Background(), hostID) + if err != nil { + t.Fatalf("get: %v", err) + } + if h.PreHookDefault != nil || h.PostHookDefault != nil { + t.Fatalf("after clear: pre=%v post=%v (want nil)", h.PreHookDefault, h.PostHookDefault) + } +} diff --git a/internal/store/hosts.go b/internal/store/hosts.go index bd6a24d..fc0b383 100644 --- a/internal/store/hosts.go +++ b/internal/store/hosts.go @@ -42,7 +42,8 @@ func (s *Store) LookupHostByAgentToken(ctx context.Context, tokenHash string) (* enrolled_at, last_seen_at, status, repo_id, tags, current_job_id, last_backup_at, last_backup_status, 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 FROM hosts WHERE agent_token_hash = ?`, tokenHash) return scanHost(row) @@ -55,7 +56,8 @@ func (s *Store) GetHost(ctx context.Context, id string) (*Host, error) { enrolled_at, last_seen_at, status, repo_id, tags, current_job_id, last_backup_at, last_backup_status, 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 FROM hosts WHERE id = ?`, id) return scanHost(row) } @@ -116,7 +118,8 @@ func (s *Store) ListHosts(ctx context.Context) ([]Host, error) { enrolled_at, last_seen_at, status, repo_id, tags, current_job_id, last_backup_at, last_backup_status, 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 FROM hosts ORDER BY name`) if err != nil { return nil, fmt.Errorf("store: list hosts: %w", err) @@ -155,13 +158,15 @@ func scanHostRow(s hostScanner) (*Host, error) { enrolled string tags string bwUp, bwDown sql.NullInt64 + preHook, postHook []byte ) err := s.Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.AgentVersion, &h.ResticVersion, &h.ProtocolVersion, &enrolled, &lastSeen, &h.Status, &repoID, &tags, ¤tJob, &lastBackupAt, &lastBkSt, &h.RepoSizeBytes, &h.SnapshotCount, &h.OpenAlertCount, - &h.AppliedScheduleVersion, &bwUp, &bwDown) + &h.AppliedScheduleVersion, &bwUp, &bwDown, + &preHook, &postHook) if err != nil { if errors.Is(err, sql.ErrNoRows) { return nil, ErrNotFound @@ -210,9 +215,24 @@ func scanHostRow(s hostScanner) (*Host, error) { v := int(bwDown.Int64) h.BandwidthDownKBps = &v } + h.PreHookDefault = preHook + h.PostHookDefault = postHook return &h, nil } +// SetHostHooks replaces the host-wide pre/post hook defaults. Pass +// nil/empty to clear that hook. Stored verbatim — caller is expected +// to encrypt the bytes before they reach this layer. +func (s *Store) SetHostHooks(ctx context.Context, hostID string, pre, post []byte) error { + _, err := s.db.ExecContext(ctx, + `UPDATE hosts SET pre_hook_default = ?, post_hook_default = ? WHERE id = ?`, + nullableBytes(pre), nullableBytes(post), hostID) + if err != nil { + return fmt.Errorf("store: set host hooks: %w", err) + } + return nil +} + // SetHostBandwidth replaces the host's upload/download caps. Pass nil // to clear a cap. Caller decides validation; non-positive caps are // treated as "no cap" by the agent regardless. diff --git a/internal/store/migrations/0010_hooks.sql b/internal/store/migrations/0010_hooks.sql new file mode 100644 index 0000000..f944dcc --- /dev/null +++ b/internal/store/migrations/0010_hooks.sql @@ -0,0 +1,25 @@ +-- 0010_hooks.sql +-- +-- P2R-10: pre/post hooks on source groups + host-wide defaults. +-- +-- Hook bodies are stored as AEAD ciphertext (existing crypto.AEAD) +-- because operators do put credentials in shell snippets — even +-- though we tell them not to. NULL means "no hook configured". +-- +-- Hooks fire only for kind=backup jobs. forget/prune/check/unlock +-- skip them per spec.md §14.3 (P2R-11 enforces this in the agent +-- dispatcher). +-- +-- Resolution order at dispatch time: +-- source_group._hook (per-group override, AEAD blob) +-- host._hook_default (host default, AEAD blob) +-- none → no hook runs +-- +-- All four columns are added in-place via ALTER TABLE ADD COLUMN. +-- Per CLAUDE.md the table-rebuild pattern is unsafe with FK cascades. + +ALTER TABLE source_groups ADD COLUMN pre_hook BLOB; +ALTER TABLE source_groups ADD COLUMN post_hook BLOB; + +ALTER TABLE hosts ADD COLUMN pre_hook_default BLOB; +ALTER TABLE hosts ADD COLUMN post_hook_default BLOB; diff --git a/internal/store/sources.go b/internal/store/sources.go index 6ec3115..164e893 100644 --- a/internal/store/sources.go +++ b/internal/store/sources.go @@ -45,13 +45,14 @@ func (st *Store) CreateSourceGroup(ctx context.Context, g *SourceGroup) error { `INSERT INTO source_groups ( id, host_id, name, includes, excludes, retention_policy, retry_max, retry_backoff_seconds, conflict_dimension, - created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + created_at, updated_at, pre_hook, post_hook + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, g.ID, g.HostID, g.Name, string(includesJSON), string(excludesJSON), string(retentionJSON), g.RetryMax, g.RetryBackoffSeconds, nullableString(g.ConflictDimension), now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), + nullableBytes(g.PreHook), nullableBytes(g.PostHook), ); err != nil { return fmt.Errorf("store: create source group: %w", err) } @@ -88,13 +89,14 @@ func (st *Store) UpdateSourceGroup(ctx context.Context, g *SourceGroup) error { `UPDATE source_groups SET name = ?, includes = ?, excludes = ?, retention_policy = ?, retry_max = ?, retry_backoff_seconds = ?, conflict_dimension = ?, - updated_at = ? + updated_at = ?, pre_hook = ?, post_hook = ? WHERE id = ? AND host_id = ?`, g.Name, string(includesJSON), string(excludesJSON), string(retentionJSON), g.RetryMax, g.RetryBackoffSeconds, nullableString(g.ConflictDimension), now.Format(time.RFC3339Nano), + nullableBytes(g.PreHook), nullableBytes(g.PostHook), g.ID, g.HostID, ) if err != nil { @@ -143,7 +145,7 @@ func (st *Store) GetSourceGroup(ctx context.Context, hostID, groupID string) (*S row := st.db.QueryRowContext(ctx, `SELECT id, host_id, name, includes, excludes, retention_policy, retry_max, retry_backoff_seconds, conflict_dimension, - created_at, updated_at + created_at, updated_at, pre_hook, post_hook FROM source_groups WHERE id = ? AND host_id = ?`, groupID, hostID) g, err := scanSourceGroup(row) @@ -159,7 +161,7 @@ func (st *Store) GetSourceGroupByName(ctx context.Context, hostID, name string) row := st.db.QueryRowContext(ctx, `SELECT id, host_id, name, includes, excludes, retention_policy, retry_max, retry_backoff_seconds, conflict_dimension, - created_at, updated_at + created_at, updated_at, pre_hook, post_hook FROM source_groups WHERE host_id = ? AND name = ?`, hostID, name) g, err := scanSourceGroup(row) @@ -177,7 +179,7 @@ func (st *Store) ListSourceGroupsByHost(ctx context.Context, hostID string) ([]S rows, err := st.db.QueryContext(ctx, `SELECT id, host_id, name, includes, excludes, retention_policy, retry_max, retry_backoff_seconds, conflict_dimension, - created_at, updated_at + created_at, updated_at, pre_hook, post_hook FROM source_groups WHERE host_id = ? ORDER BY name`, hostID) if err != nil { @@ -224,14 +226,17 @@ func scanSourceGroupRow(s sourceGroupScanner) (*SourceGroup, error) { includes, excludes, retention string conflict sql.NullString createdAt, updatedAt string + preHook, postHook []byte ) err := s.Scan(&out.ID, &out.HostID, &out.Name, &includes, &excludes, &retention, &out.RetryMax, &out.RetryBackoffSeconds, &conflict, - &createdAt, &updatedAt) + &createdAt, &updatedAt, &preHook, &postHook) if err != nil { return nil, err } + out.PreHook = preHook + out.PostHook = postHook if includes != "" { _ = json.Unmarshal([]byte(includes), &out.Includes) } @@ -259,3 +264,13 @@ func nullableString(s string) any { } return s } + +// nullableBytes returns nil for an empty/nil slice so SQL stores it +// as NULL rather than an empty BLOB. The agent treats both the same +// (no hook), but NULL is the canonical "absent" form on disk. +func nullableBytes(b []byte) any { + if len(b) == 0 { + return nil + } + return b +} diff --git a/internal/store/types.go b/internal/store/types.go index 6f99f69..4e52d91 100644 --- a/internal/store/types.go +++ b/internal/store/types.go @@ -66,6 +66,12 @@ type Host struct { // (backup, restore, prune). nil = no cap. BandwidthUpKBps *int BandwidthDownKBps *int + + // PreHookDefault / PostHookDefault are AEAD-encrypted host-wide + // hook bodies. Per source group hooks (SourceGroup.PreHook / + // PostHook) override these when set. nil = no default configured. + PreHookDefault []byte + PostHookDefault []byte } // Schedule is now intentionally slim: cron + which groups + enabled. @@ -106,6 +112,13 @@ type SourceGroup struct { ConflictDimension string CreatedAt time.Time UpdatedAt time.Time + + // PreHook / PostHook are AEAD-encrypted shell snippets (raw blob). + // nil means "no hook configured." Encryption/decryption happens at + // the HTTP layer (where AEAD lives); the store layer just persists + // the bytes verbatim. + PreHook []byte + PostHook []byte } // RetentionPolicy is the typed view of `restic forget --keep-*`. From 7b1990cf111a83ffe45d2eb7430390e0e9ff1312 Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Mon, 4 May 2026 10:57:28 +0100 Subject: [PATCH 06/15] agent+server: P2R-11 pre/post hook execution for backup jobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent: new runner.BackupHooks struct + runHook helper invoked via /bin/sh -c (cmd.exe /C on Windows). pre_hook non-zero exit aborts the backup; post_hook always runs with RM_JOB_STATUS=succeeded|failed in env. Output streamed as 'hook(): …' log.stream lines. Hooks only run for kind=backup (other kinds skip both phases). Server: resolveBackupHooks resolves group → host default → empty, decrypts via crypto.AEAD with per-slot ad bytes, plumbs plaintext into CommandRunPayload for both schedule.fire and per-group Run-now dispatch sites. Decrypt failures degrade silently to no hook so a malformed blob can't poison every backup. --- cmd/agent/main.go | 3 +- internal/agent/runner/hooks.go | 106 ++++++++++++++++++++++++++ internal/agent/runner/hooks_test.go | 90 ++++++++++++++++++++++ internal/agent/runner/runner.go | 39 +++++++++- internal/server/http/hooks_resolve.go | 75 ++++++++++++++++++ internal/server/http/run_group.go | 9 +++ internal/server/http/schedule_push.go | 14 ++++ internal/store/hooks_test.go | 30 ++++---- internal/store/hosts.go | 18 +++-- internal/store/sources.go | 24 +++--- internal/store/types.go | 23 +++--- 11 files changed, 379 insertions(+), 52 deletions(-) create mode 100644 internal/agent/runner/hooks.go create mode 100644 internal/agent/runner/hooks_test.go create mode 100644 internal/server/http/hooks_resolve.go diff --git a/cmd/agent/main.go b/cmd/agent/main.go index d6c3d8b..0c2b691 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -359,8 +359,9 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc } slog.Info("agent: accepting backup job", "job_id", p.JobID, "paths", paths, "excludes", p.Excludes, "tag", p.Tag) + hooks := runner.BackupHooks{Pre: p.PreHook, Post: p.PostHook} go func() { - if err := r.RunBackup(ctx, p.JobID, paths, p.Excludes, tags); err != nil { + if err := r.RunBackup(ctx, p.JobID, paths, p.Excludes, tags, hooks); err != nil { slog.Warn("agent: backup job failed", "job_id", p.JobID, "err", err) return } diff --git a/internal/agent/runner/hooks.go b/internal/agent/runner/hooks.go new file mode 100644 index 0000000..904b100 --- /dev/null +++ b/internal/agent/runner/hooks.go @@ -0,0 +1,106 @@ +// hooks.go — pre/post backup hooks for the agent runner (P2R-11). +// +// Hooks fire only for backup jobs (the runner's other kinds — +// init/forget/prune/check/unlock — call shell scripts that touch +// repo internals; running operator hooks for those would be +// surprising). Hook bodies arrive plaintext on the wire (server +// decrypted before the WS push). The agent never persists them +// to disk; they live in memory for the lifetime of one job. +// +// Failure semantics: +// - pre_hook non-zero exit aborts the backup: the runner returns +// the error, the job is recorded as failed, and the actual +// restic invocation never runs. +// - post_hook non-zero exit is logged with a warning prefix in +// the job log but does NOT change the job status — the operator +// wants the backup result preserved even if the cleanup step +// misbehaved. +// +// Streaming: each line of the hook's stdout/stderr is shipped as a +// log.stream envelope with payload prefixed `hook: ` so the live +// log viewer can visually separate it from restic's own output. +package runner + +import ( + "bufio" + "context" + "fmt" + "io" + "os/exec" + "runtime" + "sync/atomic" + "time" + + "gitea.dcglab.co.uk/steve/restic-manager/internal/api" +) + +// runHook executes script via the host shell. status is the value +// passed as RM_JOB_STATUS in the env (empty for pre-hooks; the +// final job status — "succeeded" or "failed" — for post-hooks). +// Returns an error iff the hook exited non-zero. ctx cancellation +// kills the subprocess. +func (r *Runner) runHook(ctx context.Context, jobID, phase, script, status string, seq *atomic.Int64) error { + if script == "" { + return nil + } + shell, flag := defaultShell() + cmd := exec.CommandContext(ctx, shell, flag, script) + cmd.Env = []string{ + "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + } + if status != "" { + cmd.Env = append(cmd.Env, "RM_JOB_STATUS="+status) + } + cmd.Env = append(cmd.Env, "RM_JOB_ID="+jobID, "RM_HOOK_PHASE="+phase) + + stdout, err := cmd.StdoutPipe() + if err != nil { + return fmt.Errorf("hook %s: stdout pipe: %w", phase, err) + } + stderr, err := cmd.StderrPipe() + if err != nil { + return fmt.Errorf("hook %s: stderr pipe: %w", phase, err) + } + if err := cmd.Start(); err != nil { + return fmt.Errorf("hook %s: start: %w", phase, err) + } + done := make(chan struct{}, 2) + go func() { r.pumpHookLines(stdout, "stdout", phase, jobID, seq); done <- struct{}{} }() + go func() { r.pumpHookLines(stderr, "stderr", phase, jobID, seq); done <- struct{}{} }() + <-done + <-done + if werr := cmd.Wait(); werr != nil { + return fmt.Errorf("hook %s exited non-zero: %w", phase, werr) + } + return nil +} + +// pumpHookLines streams lines as log.stream envelopes prefixed with +// "hook(): " so the live log can visually separate them. +func (r *Runner) pumpHookLines(rd io.Reader, stream, phase, jobID string, seq *atomic.Int64) { + scanner := bufio.NewScanner(rd) + scanner.Buffer(make([]byte, 0, 64*1024), 256*1024) + for scanner.Scan() { + line := "hook(" + phase + "): " + scanner.Text() + env, _ := api.Marshal(api.MsgLogStream, "", api.LogStreamLine{ + JobID: jobID, + Seq: seq.Add(1), + TS: time.Now().UTC(), + Stream: api.LogStream(stream), + Payload: line, + }) + _ = r.tx.Send(env) + } +} + +// defaultShell returns the (binary, single-arg-flag) pair to use for +// ` "