package api // The ZFS tab's landing-tier checks: what has waited too long on a landing branch, and verify // passes that read the pool's copy of every moved file back against the mover's ledger — // run by hand or on a schedule written to /etc/cron.d/hamlab-landing, just as snapshot // schedules are. A pass runs as its own process (`hamlab landing-verify`), the same one cron // starts, so it survives the console being restarted or upgraded mid-pass. import ( "context" "encoding/json" "io" "log" "net/http" "os" "os/exec" "path/filepath" "syscall" "time" "hamlab/internal/storage" ) func (s *Server) syncVerifyCron(ctx context.Context, z zfsState) error { exe, err := os.Executable() if err != nil { return err } exe, _ = filepath.EvalSymlinks(exe) return s.storage.WriteVerifyCron(ctx, z.Verify, s.user, exe, s.cfg.Path()) } // startVerify launches a pass in its own session: arg is "new", "all" or a schedule id. func (s *Server) startVerify(arg string) error { exe, err := os.Executable() if err != nil { return err } exe, _ = filepath.EvalSymlinks(exe) dir := filepath.Join(s.cfg.DataDir(), "landing-verify") if err := os.MkdirAll(dir, 0o700); err != nil { return err } out, err := os.OpenFile(filepath.Join(dir, "last.log"), os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) if err != nil { return err } cmd := exec.Command(exe, "-config", s.cfg.Path(), "landing-verify", arg) cmd.Stdout, cmd.Stderr = out, out cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} if err := cmd.Start(); err != nil { out.Close() return err } go func() { _ = cmd.Wait(); out.Close() }() return nil } func (s *Server) landingOverview(w http.ResponseWriter, r *http.Request) { dir := s.cfg.DataDir() now := time.Now() type tier struct { storage.LandingTarget Cursor int64 `json:"cursor"` StaleCount int `json:"stale_count"` StaleBytes int64 `json:"stale_bytes"` Stale []storage.StaleFile `json:"stale"` } tiers := []tier{} for _, t := range storage.LandingTargets() { files, n, b := storage.StaleFiles(t.Src, 100, now) if files == nil { files = []storage.StaleFile{} } tiers = append(tiers, tier{LandingTarget: t, Cursor: storage.VerifyCursor(dir, t), StaleCount: n, StaleBytes: b, Stale: files}) } z := s.zfsLoad() pols := make([]map[string]any, 0, len(z.Verify)) for i := range z.Verify { p := z.Verify[i] expr, _ := p.CronExpr() m := map[string]any{"policy": p, "cron": expr, "last": storage.LastVerify(dir, p.ID)} if p.Enabled { if n := p.NextRun(now); !n.IsZero() { m["next"] = n } } pols = append(pols, m) } hist := storage.VerifyHistory(dir) if len(hist) > 10 { hist = hist[:10] } jsonOK(w, 200, map[string]any{ "tiers": tiers, "stale_after_sec": int(storage.StaleAfter.Seconds()), "running": storage.RunningVerify(dir), "history": hist, "policies": pols, "cron_file": storage.VerifyCronFile, }) } func (s *Server) landingVerifyNow(w http.ResponseWriter, r *http.Request) { var body struct { Scope string `json:"scope"` } if err := json.NewDecoder(io.LimitReader(r.Body, 1024)).Decode(&body); err != nil || (body.Scope != "new" && body.Scope != "all") { jsonErr(w, 400, "scope: new or all") return } if storage.RunningVerify(s.cfg.DataDir()) != nil { jsonErr(w, 409, "a verify pass is already running") return } if len(storage.LandingTargets()) == 0 { jsonErr(w, 409, "no landing mover is reporting on this machine, so there is nothing to verify") return } if err := s.startVerify(body.Scope); err != nil { jsonErr(w, 500, err.Error()) return } log.Printf("landing verify (%s) started by %s", body.Scope, clientIP(r)) jsonOK(w, 202, map[string]string{"started": body.Scope}) } func (s *Server) landingPolicyPut(w http.ResponseWriter, r *http.Request) { var p storage.VerifyPolicy if err := json.NewDecoder(io.LimitReader(r.Body, 16<<10)).Decode(&p); err != nil { jsonErr(w, 400, "bad request") return } if err := p.Validate(); err != nil { jsonErr(w, 400, err.Error()) return } z := s.zfsLoad() found := false for i := range z.Verify { if p.ID != "" && z.Verify[i].ID == p.ID { z.Verify[i], found = p, true } } if !found { if p.ID == "" { id, err := randomID() if err != nil { jsonErr(w, 500, err.Error()) return } p.ID = id } z.Verify = append(z.Verify, p) } if err := s.zfsSave(z); err != nil { jsonErr(w, 500, err.Error()) return } if err := s.syncVerifyCron(r.Context(), z); err != nil { jsonErr(w, 500, "saved, but writing "+storage.VerifyCronFile+" failed: "+err.Error()) return } expr, _ := p.CronExpr() log.Printf("landing verify schedule %q saved by %s (%q, %s)", p.Name, clientIP(r), expr, p.Scope) jsonOK(w, 200, p) } func (s *Server) landingPolicyDelete(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") z := s.zfsLoad() kept := z.Verify[:0] found := false for _, p := range z.Verify { if p.ID == id { found = true continue } kept = append(kept, p) } if !found { jsonErr(w, 404, "no such schedule") return } z.Verify = kept if err := s.zfsSave(z); err != nil { jsonErr(w, 500, err.Error()) return } if err := s.syncVerifyCron(r.Context(), z); err != nil { jsonErr(w, 500, "removed, but writing "+storage.VerifyCronFile+" failed: "+err.Error()) return } storage.ForgetVerify(s.cfg.DataDir(), id) log.Printf("landing verify schedule %s removed by %s", id, clientIP(r)) w.WriteHeader(204) } func (s *Server) landingPolicyRun(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") found := false for _, p := range s.zfsLoad().Verify { found = found || p.ID == id } if !found { jsonErr(w, 404, "no such schedule") return } if storage.RunningVerify(s.cfg.DataDir()) != nil { jsonErr(w, 409, "a verify pass is already running") return } if err := s.startVerify(id); err != nil { jsonErr(w, 500, err.Error()) return } log.Printf("landing verify schedule %s run by %s", id, clientIP(r)) jsonOK(w, 202, map[string]string{"started": id}) }