diff --git a/cmd/hamlab/main.go b/cmd/hamlab/main.go index 567154c..6e8012c 100644 --- a/cmd/hamlab/main.go +++ b/cmd/hamlab/main.go @@ -63,6 +63,7 @@ func main() { passwd [user] set a user's password (makes the account if it is new; reads stdin when piped) users list the accounts zfs-snapshot run one ZFS snapshot schedule (what its cron line calls) + landing-verify verify a landing tier against its mover's ledger (a schedule's id is what its cron line calls) docker-update run one image-update schedule (what its cron line calls) user-rm remove an account and sign it out everywhere check verify the config and state load, then exit @@ -150,6 +151,10 @@ func main() { if err := zfsSnapshot(cfg, flag.Arg(1)); err != nil { log.Fatal(err) } + case "landing-verify": + if err := landingVerify(cfg, flag.Arg(1)); err != nil { + log.Fatal(err) + } case "users": migrateUsers(cfg) if len(cfg.Users) == 0 { @@ -254,6 +259,81 @@ func dockerUpdateJob(cfg *config.Config, id string) error { // // It deliberately does not write state.json. The daemon owns that file and would clobber // anything written from out here; the outcome goes beside it instead, in zfs-runs/. +// landingVerify runs one verify pass over every landing tier on this machine: by hand with a +// scope ("new" or "all"), or as a schedule by its id, which is what the cron line passes. +func landingVerify(cfg *config.Config, arg string) error { + scope, id := arg, "manual" + if arg != "new" && arg != "all" { + if strings.TrimSpace(arg) == "" { + return errors.New("usage: hamlab landing-verify ") + } + st, err := store.Open(filepath.Join(cfg.DataDir(), "state.json")) + if err != nil { + return err + } + var raw json.RawMessage + st.View(func(s *store.State) { raw = s.ZFS }) + var z struct { + Verify []storage.VerifyPolicy `json:"verify"` + } + if len(raw) > 0 { + if err := json.Unmarshal(raw, &z); err != nil { + return fmt.Errorf("reading the schedules: %w", err) + } + } + found := false + for _, p := range z.Verify { + if p.ID == arg { + scope, id, found = p.Scope, p.ID, true + } + } + if !found { + return fmt.Errorf("no verify schedule %q — it may have been removed; this cron line is stale", arg) + } + } + dir := cfg.DataDir() + unlock, err := storage.LockVerify(dir) + if err != nil { + return err + } + defer unlock() + defer storage.ClearVerifyProgress(dir) + targets := storage.LandingTargets() + if len(targets) == 0 { + r := storage.VerifyResult{ID: id, Scope: scope, Started: time.Now(), Finished: time.Now(), Issues: []storage.VerifyIssue{}, + Error: "no landing mover is reporting (nothing in " + storage.MoverStatusDir + "), so there is nothing to verify"} + _ = storage.RecordVerify(dir, storage.LandingTarget{}, r) + return errors.New(r.Error) + } + var firstErr error + for _, t := range targets { + r := storage.Verify(t, id, scope, storage.VerifyCursor(dir, t), func(p storage.VerifyProgress) { storage.WriteVerifyProgress(dir, p) }) + if err := storage.RecordVerify(dir, t, r); err != nil { + fmt.Fprintln(os.Stderr, "recording the pass:", err) + } + fmt.Printf("%s -> %s (%s): %d checked, %d match, %d mismatch, %d gone, %d unreadable, %d changed since, %d waiting, %d unhashed, %s read in %s\n", + t.Src, t.Dst, scope, r.Checked, r.Matched, r.Mismatched, r.Gone, r.Unreadable, r.Changed, r.Waiting, r.Unhashed, + humanBytes(r.BytesRead), r.Finished.Sub(r.Started).Round(time.Second)) + if r.Error != "" && firstErr == nil { + firstErr = errors.New(r.Error) + } + } + return firstErr +} + +func humanBytes(n int64) string { + const u = 1024 + if n < u { + return fmt.Sprintf("%d B", n) + } + div, exp := int64(u), 0 + for v := n / u; v >= u; v /= u { + div *= u + exp++ + } + return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp]) +} + func zfsSnapshot(cfg *config.Config, id string) error { if strings.TrimSpace(id) == "" { return errors.New("usage: hamlab zfs-snapshot ") diff --git a/internal/api/landing.go b/internal/api/landing.go new file mode 100644 index 0000000..1820e2a --- /dev/null +++ b/internal/api/landing.go @@ -0,0 +1,216 @@ +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}) +} diff --git a/internal/api/server.go b/internal/api/server.go index 8f44110..3ec2b74 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -314,6 +314,11 @@ func (s *Server) routes() *http.ServeMux { m.HandleFunc("DELETE /api/zfs/schedules/{id}", s.auth(s.zfsPolicyDelete)) m.HandleFunc("POST /api/zfs/schedules/{id}/run", s.auth(s.zfsRunNow)) m.HandleFunc("POST /api/zfs/snapshots/delete", s.auth(s.zfsSnapshotDelete)) + m.HandleFunc("GET /api/zfs/landing", s.auth(s.landingOverview)) + m.HandleFunc("POST /api/zfs/landing/verify", s.auth(s.landingVerifyNow)) + m.HandleFunc("PUT /api/zfs/landing/schedules", s.auth(s.landingPolicyPut)) + m.HandleFunc("DELETE /api/zfs/landing/schedules/{id}", s.auth(s.landingPolicyDelete)) + m.HandleFunc("POST /api/zfs/landing/schedules/{id}/run", s.auth(s.landingPolicyRun)) m.HandleFunc("GET /api/space", s.auth(s.spaceOverview)) m.HandleFunc("GET /api/space/{disk}", s.auth(s.spaceStatus)) m.HandleFunc("POST /api/space/{disk}/scan", s.auth(s.spaceScan)) diff --git a/internal/api/zfs.go b/internal/api/zfs.go index 5b9e5ae..2cea987 100644 --- a/internal/api/zfs.go +++ b/internal/api/zfs.go @@ -26,7 +26,8 @@ import ( ) type zfsState struct { - Policies []storage.Policy `json:"policies"` + Policies []storage.Policy `json:"policies"` + Verify []storage.VerifyPolicy `json:"verify"` // landing-tier verify schedules (landing.go) } func (s *Server) zfsLoad() zfsState { @@ -39,6 +40,9 @@ func (s *Server) zfsLoad() zfsState { if z.Policies == nil { z.Policies = []storage.Policy{} } + if z.Verify == nil { + z.Verify = []storage.VerifyPolicy{} + } return z } diff --git a/internal/storage/landing.go b/internal/storage/landing.go new file mode 100644 index 0000000..6c6b243 --- /dev/null +++ b/internal/storage/landing.go @@ -0,0 +1,564 @@ +package storage + +// Verifying a landing tier: did everything that landed really reach the pool, intact? +// +// A landing tier is fast storage in front of a pool (a mergerfs pool, new files on the fast +// branch first) with a mover draining it. The mover keeps a ledger — one JSON line per file it +// moved: the path under the pool, its size and mtime, and the SHA-256 of the fast copy read +// just before the move (hamlet's scripts/landing-mover.py writes /var/lib/landing-mover/ +// ledger.jsonl and names it in its status file). A verify pass reads the pool's copy of each +// entry back and compares. +// +// What a pass can say about each file: +// - match: same size, same mtime, same SHA-256 — it arrived intact +// - mismatch: same size and mtime but different contents — the one real failure +// - changed: the pool's copy has a different size or mtime: edited since it moved, so the +// old hash says nothing about it (counted, not compared) +// - gone: not on the pool and not waiting on the landing branch either — deleted or +// renamed since it moved. A pass cannot tell that from a loss, so it lists them +// - waiting: back on the landing branch (written again since), the mover will take it +// - unreadable: could not be opened as this account +// - unhashed: the mover recorded no hash for it (not a regular file, or changed mid-move) +// +// Scope "new" checks what has moved since the last pass, "all" re-reads the whole ledger. A +// path moved more than once is checked against its latest entry only. + +import ( + "bufio" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + "syscall" + "time" +) + +// ---- schedules ---- + +// VerifyPolicy runs a verify pass on a schedule. The when-fields mean what they do on a +// snapshot Policy; Scope is "new" or "all". +type VerifyPolicy struct { + ID string `json:"id"` + Name string `json:"name"` + Every string `json:"every"` + At string `json:"at"` + Weekday int `json:"weekday"` + Day int `json:"day"` + Cron string `json:"cron,omitempty"` + Scope string `json:"scope"` + Enabled bool `json:"enabled"` +} + +// when borrows the snapshot Policy's schedule logic, so both kinds of schedule are written, +// validated and read back by exactly the same code. +func (v *VerifyPolicy) when() *Policy { + return &Policy{Every: v.Every, At: v.At, Weekday: v.Weekday, Day: v.Day, Cron: v.Cron} +} + +func (v *VerifyPolicy) CronExpr() (string, error) { return v.when().CronExpr() } +func (v *VerifyPolicy) NextRun(now time.Time) time.Time { return v.when().NextRun(now) } + +func (v *VerifyPolicy) Validate() error { + if v.Scope != "new" && v.Scope != "all" { + return errors.New("scope: new (what moved since the last pass) or all (the whole ledger)") + } + switch v.Every { + case "hourly", "daily", "weekly", "monthly": + case "custom": + if strings.TrimSpace(v.Cron) == "" { + return errors.New("a custom schedule needs a cron expression") + } + default: + return errors.New("every: hourly, daily, weekly, monthly or custom") + } + if _, err := v.CronExpr(); err != nil { + return err + } + if _, _, err := v.when().clock(); err != nil { + return err + } + if v.Every == "weekly" && (v.Weekday < 0 || v.Weekday > 6) { + return errors.New("weekday: 0 (Sunday) to 6") + } + if v.Every == "monthly" && (v.Day < 1 || v.Day > 28) { + return errors.New("day: 1 to 28, so every month has one") + } + if len(v.Name) > 64 { + return errors.New("name: 64 characters at most") + } + return nil +} + +// VerifyCronFile holds the verify schedules; Ham Lab owns it and rewrites it whole. +const VerifyCronFile = "/etc/cron.d/hamlab-landing" + +// VerifyCronText renders the verify schedules as cron lines, run as Ham Lab's own account. +func VerifyCronText(ps []VerifyPolicy, user, exe, config string) (string, error) { + var b strings.Builder + b.WriteString("# Ham Lab landing-tier verify schedules — written by the ZFS tab, rewritten whole on every change.\n") + b.WriteString("# Edit them there, not here: anything added by hand is lost the next time a schedule changes.\n") + b.WriteString("SHELL=/bin/sh\n") + b.WriteString("PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin\n\n") + for i := range ps { + p := &ps[i] + name := strings.ReplaceAll(p.Name, "\n", " ") + if !p.Enabled { + b.WriteString("# " + name + " — paused\n") + continue + } + expr, err := p.CronExpr() + if err != nil { + return "", fmt.Errorf("%s: %w", p.Name, err) + } + fmt.Fprintf(&b, "# %s — verify %s\n", name, p.Scope) + fmt.Fprintf(&b, "%s %s %s -config %s landing-verify %s\n\n", expr, user, exe, config, p.ID) + } + return b.String(), nil +} + +// WriteVerifyCron puts the verify schedules on the machine, or removes the file when none run. +func (m *Manager) WriteVerifyCron(ctx context.Context, ps []VerifyPolicy, user, exe, config string) error { + live := false + for i := range ps { + live = live || ps[i].Enabled + } + if !live { + return m.RemoveRootFile(ctx, VerifyCronFile) + } + text, err := VerifyCronText(ps, user, exe, config) + if err != nil { + return err + } + return m.writeRoot(ctx, VerifyCronFile, text) +} + +// ---- finding the landing tier ---- + +// MoverStatusDir is where a landing mover writes what it is doing. +const MoverStatusDir = "/run/landing-mover" + +// DefaultLedger is where hamlet's mover keeps its ledger when no status file names one. +const DefaultLedger = "/var/lib/landing-mover/ledger.jsonl" + +// LandingTarget is one landing tier a pass can check: the branch files land on, the pool +// directory they are moved to, and the mover's ledger. +type LandingTarget struct { + Src string `json:"src"` + Dst string `json:"dst"` + Ledger string `json:"ledger"` +} + +// LandingTargets reads the movers' status files. Each mover names its own source, target and +// ledger; with none running, nothing can be verified, and the page says so. +func LandingTargets() []LandingTarget { + var out []LandingTarget + m, _ := filepath.Glob(MoverStatusDir + "/*.json") + for _, p := range m { + b, err := os.ReadFile(p) + if err != nil { + continue + } + var st struct{ Src, Dst, Ledger string } + if json.Unmarshal(b, &st) != nil || st.Src == "" || st.Dst == "" { + continue + } + if st.Ledger == "" { + st.Ledger = DefaultLedger + } + out = append(out, LandingTarget{Src: filepath.Clean(st.Src), Dst: filepath.Clean(st.Dst), Ledger: st.Ledger}) + } + sort.Slice(out, func(i, j int) bool { return out[i].Src < out[j].Src }) + return out +} + +// ---- stale files ---- + +// StaleAfter is how long a file may wait on a landing branch before it counts as stale. A file +// older than this is not necessarily lost — a big backlog drains for hours — but it is worth a +// look, which is what the ZFS tab offers. +const StaleAfter = time.Hour + +// StaleFile is one file that has waited longer than StaleAfter. +type StaleFile struct { + Path string `json:"path"` // relative to the landing branch, which is also its path on the pool + Size int64 `json:"size"` + MTime int64 `json:"mtime"` // unix ms +} + +// StaleFiles lists what has waited on a landing branch longer than StaleAfter, oldest first, +// at most limit of them, with the full count and size. +func StaleFiles(branch string, limit int, now time.Time) (files []StaleFile, count int, bytes int64) { + cut := now.Add(-StaleAfter) + _ = filepath.WalkDir(branch, func(p string, d os.DirEntry, err error) error { + if err != nil || !d.Type().IsRegular() { + return nil + } + fi, err := d.Info() + if err != nil || !fi.ModTime().Before(cut) { + return nil + } + count++ + bytes += fi.Size() + rel, _ := filepath.Rel(branch, p) + files = append(files, StaleFile{Path: rel, Size: fi.Size(), MTime: fi.ModTime().UnixMilli()}) + return nil + }) + sort.Slice(files, func(i, j int) bool { return files[i].MTime < files[j].MTime }) + if len(files) > limit { + files = files[:limit] + } + return files, count, bytes +} + +// ---- the pass ---- + +// LedgerEntry is one line of the mover's ledger. +type LedgerEntry struct { + T int64 `json:"t"` // when it was moved, unix ms + Path string `json:"path"` + Size int64 `json:"size"` + MTimeNs int64 `json:"mtime_ns"` + SHA256 string `json:"sha256,omitempty"` + Note string `json:"note,omitempty"` +} + +// VerifyIssue is one file a pass wants a person to look at. +type VerifyIssue struct { + Path string `json:"path"` + Kind string `json:"kind"` // mismatch | gone | unreadable + Moved int64 `json:"moved"` + Size int64 `json:"size"` + Note string `json:"note,omitempty"` +} + +// VerifyResult is what one pass found. +type VerifyResult struct { + ID string `json:"id"` // the schedule that ran it, or "manual" + Scope string `json:"scope"` + Started time.Time `json:"started"` + Finished time.Time `json:"finished"` + Src string `json:"src"` + Dst string `json:"dst"` + Since int64 `json:"since"` // ledger entries after this (unix ms); 0 for all + Entries int `json:"entries"` + Checked int `json:"checked"` + Matched int `json:"matched"` + Mismatched int `json:"mismatched"` + Changed int `json:"changed"` + Gone int `json:"gone"` + Waiting int `json:"waiting"` + Unreadable int `json:"unreadable"` + Unhashed int `json:"unhashed"` + BytesRead int64 `json:"bytes_read"` + Issues []VerifyIssue `json:"issues"` + Error string `json:"error,omitempty"` + Cursor int64 `json:"cursor"` // the newest entry it covered +} + +// OK is true when nothing needs a look: no mismatches, nothing gone or unreadable, no error. +func (r *VerifyResult) OK() bool { + return r.Error == "" && r.Mismatched == 0 && r.Gone == 0 && r.Unreadable == 0 +} + +// VerifyProgress is published while a pass runs. +type VerifyProgress struct { + ID string `json:"id"` + Scope string `json:"scope"` + PID int `json:"pid"` + Started time.Time `json:"started"` + Updated time.Time `json:"updated"` + Total int `json:"total"` + Done int `json:"done"` + BytesRead int64 `json:"bytes_read"` + BytesAll int64 `json:"bytes_all"` + Current string `json:"current"` +} + +const maxIssues = 500 + +// readLedger loads the ledger's entries newer than since, the latest per path, oldest first. +func readLedger(path string, since int64) ([]LedgerEntry, error) { + f, err := os.Open(path) + if err != nil { + return nil, err + } + defer f.Close() + latest := map[string]LedgerEntry{} + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 64<<10), 4<<20) + for sc.Scan() { + var e LedgerEntry + if json.Unmarshal(sc.Bytes(), &e) != nil || e.Path == "" || e.T <= since { + continue + } + if old, ok := latest[e.Path]; !ok || e.T >= old.T { + latest[e.Path] = e + } + } + if err := sc.Err(); err != nil { + return nil, err + } + out := make([]LedgerEntry, 0, len(latest)) + for _, e := range latest { + out = append(out, e) + } + sort.Slice(out, func(i, j int) bool { return out[i].T < out[j].T }) + return out, nil +} + +// inside joins rel under root and refuses anything that would climb out of it. +func inside(root, rel string) (string, bool) { + p := filepath.Join(root, rel) + return p, p == root || strings.HasPrefix(p, root+string(filepath.Separator)) +} + +func hashFile(p string, buf []byte, onRead func(int)) (string, error) { + f, err := os.Open(p) + if err != nil { + return "", err + } + defer f.Close() + h := sha256.New() + for { + n, err := f.Read(buf) + if n > 0 { + h.Write(buf[:n]) + onRead(n) + } + if err == io.EOF { + return hex.EncodeToString(h.Sum(nil)), nil + } + if err != nil { + return "", err + } + } +} + +// Verify runs one pass over a landing tier's ledger. progress is called about once a second. +func Verify(t LandingTarget, id, scope string, since int64, progress func(VerifyProgress)) VerifyResult { + r := VerifyResult{ID: id, Scope: scope, Started: time.Now(), Src: t.Src, Dst: t.Dst, Since: since, Issues: []VerifyIssue{}, Cursor: since} + if scope == "all" { + r.Since = 0 + } + entries, err := readLedger(t.Ledger, r.Since) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + r.Error = "no ledger yet at " + t.Ledger + " — nothing has been moved since the mover started keeping one" + } else { + r.Error = "reading the ledger: " + err.Error() + } + r.Finished = time.Now() + return r + } + r.Entries = len(entries) + pr := VerifyProgress{ID: id, Scope: scope, PID: os.Getpid(), Started: r.Started, Total: len(entries)} + for _, e := range entries { + pr.BytesAll += e.Size + } + last := time.Time{} + tick := func(force bool) { + if progress != nil && (force || time.Since(last) >= time.Second) { + pr.Updated, last = time.Now(), time.Now() + progress(pr) + } + } + issue := func(e LedgerEntry, kind, note string) { + if len(r.Issues) < maxIssues { + r.Issues = append(r.Issues, VerifyIssue{Path: e.Path, Kind: kind, Moved: e.T, Size: e.Size, Note: note}) + } + } + buf := make([]byte, 4<<20) + for _, e := range entries { + pr.Current = e.Path + tick(false) + if e.T > r.Cursor { + r.Cursor = e.T + } + r.Checked++ + pr.Done++ + dst, ok1 := inside(t.Dst, e.Path) + src, ok2 := inside(t.Src, e.Path) + if !ok1 || !ok2 { + r.Unreadable++ + issue(e, "unreadable", "path climbs out of the pool") + continue + } + st, err := os.Lstat(dst) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + if _, err := os.Lstat(src); err == nil { + r.Waiting++ + continue + } + r.Gone++ + issue(e, "gone", "not on the pool, not waiting to move: deleted or renamed since it moved?") + continue + } + r.Unreadable++ + issue(e, "unreadable", err.Error()) + continue + } + if e.SHA256 == "" { + r.Unhashed++ + continue + } + mt := st.ModTime().UnixNano() + if st.Size() != e.Size || abs64(mt-e.MTimeNs) >= int64(time.Second) { + r.Changed++ + continue + } + sum, err := hashFile(dst, buf, func(n int) { r.BytesRead += int64(n); pr.BytesRead += int64(n); tick(false) }) + if err != nil { + if errors.Is(err, syscall.EACCES) || errors.Is(err, os.ErrPermission) { + r.Unreadable++ + issue(e, "unreadable", "permission denied") + } else { + r.Unreadable++ + issue(e, "unreadable", err.Error()) + } + continue + } + if sum != e.SHA256 { + r.Mismatched++ + issue(e, "mismatch", "contents differ from the copy that landed") + continue + } + r.Matched++ + } + tick(true) + r.Finished = time.Now() + return r +} + +func abs64(v int64) int64 { + if v < 0 { + return -v + } + return v +} + +// ---- records ---- + +func verifyDir(dataDir string) string { return filepath.Join(dataDir, "landing-verify") } + +func writeJSON(p string, v any) error { + if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil { + return err + } + b, err := json.Marshal(v) + if err != nil { + return err + } + tmp := p + ".tmp" + if err := os.WriteFile(tmp, b, 0o600); err != nil { + return err + } + return os.Rename(tmp, p) +} + +func readJSON(p string, v any) bool { + b, err := os.ReadFile(p) + return err == nil && json.Unmarshal(b, v) == nil +} + +// cursorKey names a landing tier's cursor file. +func cursorKey(t LandingTarget) string { + return strings.Trim(strings.NewReplacer("/", "_").Replace(t.Src), "_") +} + +// VerifyCursor is the newest ledger entry a finished pass covered, for scope "new". +func VerifyCursor(dataDir string, t LandingTarget) int64 { + var c struct{ Cursor int64 } + readJSON(filepath.Join(verifyDir(dataDir), "cursor-"+cursorKey(t)+".json"), &c) + return c.Cursor +} + +// RecordVerify keeps a finished pass: as the run's last result, in the history (newest first, +// 30 kept), and — when it read the ledger at all — as the new cursor. +func RecordVerify(dataDir string, t LandingTarget, r VerifyResult) error { + d := verifyDir(dataDir) + if err := writeJSON(filepath.Join(d, "run-"+r.ID+".json"), r); err != nil { + return err + } + var hist []VerifyResult + readJSON(filepath.Join(d, "history.json"), &hist) + hist = append([]VerifyResult{r}, hist...) + if len(hist) > 30 { + hist = hist[:30] + } + if err := writeJSON(filepath.Join(d, "history.json"), hist); err != nil { + return err + } + if r.Error == "" && r.Cursor > VerifyCursor(dataDir, t) { + return writeJSON(filepath.Join(d, "cursor-"+cursorKey(t)+".json"), map[string]int64{"cursor": r.Cursor}) + } + return nil +} + +// VerifyHistory is the kept passes, newest first. +func VerifyHistory(dataDir string) []VerifyResult { + var hist []VerifyResult + readJSON(filepath.Join(verifyDir(dataDir), "history.json"), &hist) + if hist == nil { + hist = []VerifyResult{} + } + return hist +} + +// LastVerify is the last pass a schedule (or "manual") ran. +func LastVerify(dataDir, id string) *VerifyResult { + var r VerifyResult + if !readJSON(filepath.Join(verifyDir(dataDir), "run-"+id+".json"), &r) { + return nil + } + return &r +} + +// ForgetVerify drops a removed schedule's last result (the history keeps it). +func ForgetVerify(dataDir, id string) { + _ = os.Remove(filepath.Join(verifyDir(dataDir), "run-"+id+".json")) +} + +func progressPath(dataDir string) string { return filepath.Join(verifyDir(dataDir), "running.json") } + +// WriteVerifyProgress publishes a running pass. +func WriteVerifyProgress(dataDir string, p VerifyProgress) { _ = writeJSON(progressPath(dataDir), p) } + +// ClearVerifyProgress removes it once the pass ends. +func ClearVerifyProgress(dataDir string) { _ = os.Remove(progressPath(dataDir)) } + +// RunningVerify is the pass in progress, or nil. A progress file whose process is gone, or +// that has not moved for two minutes, is a pass that died: it is reported as not running. +func RunningVerify(dataDir string) *VerifyProgress { + var p VerifyProgress + if !readJSON(progressPath(dataDir), &p) { + return nil + } + if p.PID <= 0 || syscall.Kill(p.PID, 0) != nil || time.Since(p.Updated) > 2*time.Minute { + return nil + } + return &p +} + +// LockVerify takes the one-pass-at-a-time lock; release it with the returned func. +func LockVerify(dataDir string) (func(), error) { + if err := os.MkdirAll(verifyDir(dataDir), 0o700); err != nil { + return nil, err + } + f, err := os.OpenFile(filepath.Join(verifyDir(dataDir), "lock"), os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, err + } + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + f.Close() + return nil, errors.New("a verify pass is already running") + } + return func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN); f.Close() }, nil +} diff --git a/internal/storage/landing_test.go b/internal/storage/landing_test.go new file mode 100644 index 0000000..6a34fe1 --- /dev/null +++ b/internal/storage/landing_test.go @@ -0,0 +1,137 @@ +package storage + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestVerifyClassifiesEveryCase(t *testing.T) { + root := t.TempDir() + src, dst := filepath.Join(root, "landing"), filepath.Join(root, "pool") + for _, d := range []string{src, dst + "/a", src + "/a"} { + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatal(err) + } + } + mtime := time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC) + put := func(p, body string) LedgerEntry { + if err := os.WriteFile(p, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + _ = os.Chtimes(p, mtime, mtime) + sum := sha256.Sum256([]byte(body)) + return LedgerEntry{Size: int64(len(body)), MTimeNs: mtime.UnixNano(), SHA256: hex.EncodeToString(sum[:])} + } + var ledger []LedgerEntry + add := func(rel string, e LedgerEntry, at int64) { e.Path, e.T = rel, at; ledger = append(ledger, e) } + + add("a/match.bin", put(dst+"/a/match.bin", "intact"), 10) + e := put(dst+"/a/mismatch.bin", "garbled") // same size and mtime as what was hashed, other bytes + sum := sha256.Sum256([]byte("GARBLED")) + e.SHA256 = hex.EncodeToString(sum[:]) + add("a/mismatch.bin", e, 11) + e = put(dst+"/a/changed.bin", "edited since, longer now") + e.Size = 3 + add("a/changed.bin", e, 12) + add("a/gone.bin", LedgerEntry{Size: 4, MTimeNs: mtime.UnixNano(), SHA256: "00"}, 13) + put(src+"/a/waiting.bin", "back again") + add("a/waiting.bin", LedgerEntry{Size: 10, MTimeNs: mtime.UnixNano(), SHA256: "00"}, 14) + e = put(dst+"/a/unhashed", "x") + e.SHA256, e.Note = "", "not a regular file" + add("a/unhashed", e, 15) + add("../escape", LedgerEntry{Size: 1, SHA256: "00"}, 16) + // moved twice: only the newer entry counts, and it matches + add("a/twice.bin", LedgerEntry{Size: 99, SHA256: "stale"}, 5) + add("a/twice.bin", put(dst+"/a/twice.bin", "second time"), 17) + + lp := filepath.Join(root, "ledger.jsonl") + f, _ := os.Create(lp) + for _, e := range ledger { + b, _ := json.Marshal(e) + f.Write(append(b, '\n')) + } + f.WriteString("not json\n") + f.Close() + + tg := LandingTarget{Src: src, Dst: dst, Ledger: lp} + r := Verify(tg, "manual", "all", 0, nil) + want := map[string]int{"checked": 8, "matched": 2, "mismatched": 1, "changed": 1, "gone": 1, "waiting": 1, "unhashed": 1, "unreadable": 1} + got := map[string]int{"checked": r.Checked, "matched": r.Matched, "mismatched": r.Mismatched, "changed": r.Changed, + "gone": r.Gone, "waiting": r.Waiting, "unhashed": r.Unhashed, "unreadable": r.Unreadable} + for k, v := range want { + if got[k] != v { + t.Errorf("%s: got %d want %d (%+v)", k, got[k], v, r) + } + } + if r.OK() || r.Cursor != 17 || r.Error != "" { + t.Errorf("ok=%v cursor=%d err=%q", r.OK(), r.Cursor, r.Error) + } + kinds := map[string]string{} + for _, i := range r.Issues { + kinds[i.Path] = i.Kind + } + if kinds["a/mismatch.bin"] != "mismatch" || kinds["a/gone.bin"] != "gone" || kinds["../escape"] != "unreadable" { + t.Errorf("issues %+v", r.Issues) + } + + // "new" after a pass at cursor 14 only looks at what moved later + r = Verify(tg, "manual", "new", 14, nil) + if r.Checked != 3 || r.Matched != 1 || r.Unhashed != 1 || r.Unreadable != 1 { + t.Errorf("new since 14: %+v", r) + } + + // records: the cursor only advances on a pass that read the ledger + dir := t.TempDir() + if err := RecordVerify(dir, tg, Verify(tg, "manual", "all", 0, nil)); err != nil { + t.Fatal(err) + } + if c := VerifyCursor(dir, tg); c != 17 { + t.Errorf("cursor %d", c) + } + bad := Verify(LandingTarget{Src: src, Dst: dst, Ledger: filepath.Join(root, "missing")}, "manual", "new", 0, nil) + if !strings.Contains(bad.Error, "no ledger yet") { + t.Errorf("missing ledger: %q", bad.Error) + } + if len(VerifyHistory(dir)) != 1 || LastVerify(dir, "manual") == nil { + t.Errorf("history/last not kept") + } +} + +func TestVerifyPolicy(t *testing.T) { + p := VerifyPolicy{ID: "abc", Name: "nightly", Every: "daily", At: "05:10", Scope: "new", Enabled: true} + if err := p.Validate(); err != nil { + t.Fatal(err) + } + if x, _ := p.CronExpr(); x != "10 5 * * *" { + t.Fatalf("cron %q", x) + } + txt, err := VerifyCronText([]VerifyPolicy{p, {ID: "off", Name: "paused one", Every: "hourly", Scope: "all"}}, "core", "/usr/local/bin/hamlab", "/cfg.json") + if err != nil || !strings.Contains(txt, "10 5 * * * core /usr/local/bin/hamlab -config /cfg.json landing-verify abc") || !strings.Contains(txt, "# paused one — paused") { + t.Fatalf("%v\n%s", err, txt) + } + for _, b := range []VerifyPolicy{{Every: "daily", Scope: "some"}, {Every: "sometimes", Scope: "new"}, {Every: "custom", Scope: "new"}} { + if b.Validate() == nil { + t.Errorf("accepted %+v", b) + } + } +} + +func TestStaleFiles(t *testing.T) { + d := t.TempDir() + now := time.Now() + os.MkdirAll(d+"/x", 0o755) + for name, age := range map[string]time.Duration{"x/old.bin": 3 * time.Hour, "x/older.bin": 5 * time.Hour, "fresh.bin": time.Minute} { + os.WriteFile(filepath.Join(d, name), []byte("12345"), 0o644) + os.Chtimes(filepath.Join(d, name), now.Add(-age), now.Add(-age)) + } + files, n, b := StaleFiles(d, 1, now) + if n != 2 || b != 10 || len(files) != 1 || files[0].Path != "x/older.bin" { + t.Fatalf("%d %d %+v", n, b, files) + } +} diff --git a/web/src/ZFS.tsx b/web/src/ZFS.tsx index 4a42b32..c2e812b 100644 --- a/web/src/ZFS.tsx +++ b/web/src/ZFS.tsx @@ -9,7 +9,7 @@ // disagree over. An existing cron job is detected and shown, so the page tells the truth // about the machine whether Ham Lab or something else is doing the work. import { FormEvent, Fragment, useCallback, useEffect, useState } from 'react' -import { api, ARCView, CacheDev, Landing, PoolCaching, PoolWrites, ZFSExternal, ZFSInfo, ZFSOverview, ZFSPolicy, ZFSRun, ZFSSnapshot, ZFSWrites } from './api' +import { api, ARCView, CacheDev, Landing, LandingCheck, PoolCaching, PoolWrites, VerifyPolicy, VerifyResult, ZFSExternal, ZFSInfo, ZFSOverview, ZFSPolicy, ZFSRun, ZFSSnapshot, ZFSWrites } from './api' import { fmtBytes, fmtRate } from './fmt' import { useRefreshMs } from './prefs' import { fetchHistories } from './sensors' @@ -588,6 +588,253 @@ function ARCSection() { ) } +// ---- landing-tier checks: what has waited too long, and verify passes ---- +// +// A verify pass reads the pool's copy of every file the mover moved back against the SHA-256 +// the mover took of the landing copy, from its ledger. "New moves" checks what has moved since +// the last pass; "everything" re-reads the whole ledger. Passes run as their own process, the +// same one a schedule's cron line starts, so a console restart does not cut one short. + +const vDescribe = (p: VerifyPolicy) => { + const when = p.every === 'custom' ? `on ${p.cron}` + : p.every === 'hourly' ? `every hour at :${(p.at.split(':')[1] ?? '00').padStart(2, '0')}` + : p.every === 'daily' ? `every day at ${p.at}` + : p.every === 'weekly' ? `every ${WEEKDAYS[p.weekday] ?? 'Sunday'} at ${p.at}` + : `on day ${p.day} of the month at ${p.at}` + return `${when} · ${p.scope === 'all' ? 'everything in the ledger' : 'what moved since the last pass'}` +} +const vOK = (r: VerifyResult) => !r.error && r.mismatched === 0 && r.gone === 0 && r.unreadable === 0 +const secs = (a: string, b: string) => (new Date(b).getTime() - new Date(a).getTime()) / 1000 + +function VerifySummary({ r }: { r: VerifyResult }) { + if (r.error) return {r.error} + const extra = [r.changed && `${fmtN(r.changed)} edited since (not compared)`, r.waiting && `${fmtN(r.waiting)} waiting to move again`, r.unhashed && `${fmtN(r.unhashed)} without a hash`].filter(Boolean).join(' · ') + return ( + + {vOK(r) + ? <>{r.checked === 0 ? 'nothing new to check' : <>{fmtN(r.matched)} of {fmtN(r.checked)} files match} + : {[r.mismatched && `${fmtN(r.mismatched)} DIFFER`, r.gone && `${fmtN(r.gone)} not on the pool`, r.unreadable && `${fmtN(r.unreadable)} unreadable`].filter(Boolean).join(' · ')} · {fmtN(r.matched)} match} + · {fmtBytes(r.bytes_read)} read in {dur(secs(r.started, r.finished)) === '—' ? '0 s' : dur(secs(r.started, r.finished))} · {r.scope === 'all' ? 'everything' : 'new moves'} · {ago(r.finished)} + {extra && · {extra}} + + ) +} + +function Issues({ r }: { r: VerifyResult }) { + const [open, setOpen] = useState(false) + if (!r.issues?.length) return null + return ( +
+ + {open &&
{r.issues.map((i) => ( +
+ {i.kind === 'mismatch' ? 'differs' : i.kind === 'gone' ? 'not on pool' : 'unreadable'} + {i.path} + moved {new Date(i.moved).toLocaleString()} + {fmtBytes(i.size)} +
+ ))}
} + {open && r.gone > 0 &&
"Not on the pool" is a file that was moved and is no longer there — deleted or renamed since, most often. The pass cannot tell that from a loss, so it lists them.
} +
+ ) +} + +function VerifyScheduleCard({ p, cron, next, last, onChanged, onEdit }: { p: VerifyPolicy; cron: string; next?: string; last: VerifyResult | null; onChanged: () => void; onEdit: () => void }) { + const run = async () => { + try { await api.runVerifySchedule(p.id); toast('verify pass started'); onChanged() } catch (e) { toast(e instanceof Error ? e.message : 'failed', 'err') } + } + const remove = async () => { + if (!await askYes(`Remove the verify schedule "${p.name}"? Passes it already ran stay in the history.`, 'Remove')) return + try { await api.deleteVerifySchedule(p.id); toast('schedule removed'); onChanged() } catch (e) { toast(e instanceof Error ? e.message : 'failed', 'err') } + } + const toggle = async () => { + try { await api.saveVerifySchedule({ ...p, enabled: !p.enabled }); onChanged() } catch (e) { toast(e instanceof Error ? e.message : 'failed', 'err') } + } + return ( +
+
+ + {p.name || 'verify'} + {!p.enabled && paused} + + {p.enabled ? soon(next) : 'not scheduled'} +
+
+
When{vDescribe(p)}
+
Cron{cron}
+
Last pass{last ? : never run}
+
+
+ Run now + + {p.enabled ? 'Pause' : 'Resume'} + Remove +
+
+ ) +} + +const VBLANK: Partial = { name: '', every: 'daily', at: '05:10', weekday: 0, day: 1, scope: 'new', enabled: true } + +function VerifyDialog({ p, onClose, onSaved }: { p: Partial; onClose: () => void; onSaved: () => void }) { + const [f, setF] = useState>({ ...VBLANK, ...p }) + const [err, setErr] = useState('') + useEscape(onClose) + const set = (k: keyof VerifyPolicy, v: unknown) => setF((x) => ({ ...x, [k]: v })) + const save = async (e: FormEvent) => { + e.preventDefault(); setErr('') + try { await api.saveVerifySchedule({ ...f, name: (f.name || '').trim() || (f.scope === 'all' ? 'verify everything' : 'verify new moves') }); onSaved() } + catch (e) { setErr(e instanceof Error ? e.message : 'failed') } + } + return ( +
+
e.stopPropagation()} onSubmit={save}> +

{f.id ? 'Edit verify schedule' : 'New verify schedule'}

+ + + + {f.every === 'custom' ? ( + + ) : ( + + )} + {f.every === 'weekly' && ( + + )} + {f.every === 'monthly' && ( + + )} + + {err &&
{err}
} +
+ + + +
+
+
+ ) +} + +function LandingChecks() { + const [c, setC] = useState(null) + const [err, setErr] = useState('') + const [showStale, setShowStale] = useState(false) + const [editing, setEditing] = useState | null>(null) + const rate = useRefreshMs() + const load = useCallback(async () => { + try { setC(await api.zfsLanding()); setErr('') } catch (e) { setErr(e instanceof Error ? e.message : 'failed') } + }, []) + useEffect(() => { load() }, [load]) + const running = !!c?.running + useEffect(() => { const t = setInterval(load, running ? 2000 : Math.max(10000, rate * 5)); return () => clearInterval(t) }, [load, rate, running]) + + if (err && !c) return
{err}
+ if (!c || (c.tiers.length === 0 && c.policies.length === 0 && c.history.length === 0)) return null + const verify = async (scope: 'new' | 'all') => { + if (scope === 'all' && !await askYes('Re-read every file the mover has ever moved, from the pool, and check each against its hash? It reads as much as the ledger holds.', 'Verify everything')) return + try { await api.landingVerify(scope); toast('verify pass started'); setTimeout(load, 800) } catch (e) { toast(e instanceof Error ? e.message : 'failed', 'err') } + } + const last = c.history[0] + const hours = Math.round(c.stale_after_sec / 3600) + const r = c.running + return ( + <> +

Landing tier checks

+ did everything that landed reach the pool, intact? +
+
+ {c.tiers.length === 0 &&
No landing mover is reporting right now (nothing in /run/landing-mover/), so there is nothing to check. Past passes are below.
} + {c.tiers.map((t) => ( +
+
+ + {t.dst} + landing checks + + {t.src} → {t.dst} +
+
+
Stale + {t.stale_count + ? <>{fmtN(t.stale_count)} file{t.stale_count === 1 ? '' : 's'}, {fmtBytes(t.stale_bytes)}, waiting over {hours} h{' '} + : `nothing has waited on the landing tier over ${hours} h`} +
+
Last pass{last ? : never verified}
+
Covered{t.cursor ? `every file moved up to ${new Date(t.cursor).toLocaleString()} has been checked at least once` : 'nothing checked yet'} · ledger {t.ledger}
+
+ {showStale && t.stale.length > 0 && ( +
{t.stale.map((f) => ( +
+ {f.path} + landed {new Date(f.mtime).toLocaleString()} + waiting {dur((Date.now() - f.mtime) / 1000)} + {fmtBytes(f.size)} +
+ ))}{t.stale_count > t.stale.length &&
and {fmtN(t.stale_count - t.stale.length)} more
}
+ )} + {r && ( +
+
Verifying {r.scope === 'all' ? 'everything' : 'new moves'} · {fmtN(r.done)} of {fmtN(r.total)} files · {fmtBytes(r.bytes_read)} of {fmtBytes(r.bytes_all)}{r.current}
+ 0 ? r.bytes_read / r.bytes_all : r.total > 0 ? r.done / r.total : 0} cls="" /> +
+ )} + {last && } +
+ verify('new')}>{running ? 'Verifying…' : 'Verify new moves'} + verify('all')}>Verify everything +
+ {c.history.length > 1 && ( +
+ Earlier passes +
{c.history.slice(1).map((h, i) => ( +
{new Date(h.finished).toLocaleString()}
+ ))}
+
+ )} +
+ ))} +
+ +

Verify schedules

+ setEditing({ ...VBLANK })}>+ New schedule +
+ {c.policies.length === 0 &&
No verify runs on a schedule. A daily pass of new moves checks every file once, soon after it lands.
} +
{c.policies.map(({ policy, cron, next, last }) => ( + setEditing(policy)} /> + ))}
+ {editing && setEditing(null)} onSaved={() => { setEditing(null); load() }} />} + + ) +} + // ---- page ---- export default function ZFS() { @@ -620,6 +867,7 @@ export default function ZFS() { the ARC in RAM, and what each pool has in front of its disks +

Snapshot schedules

setEditing({ ...BLANK, dataset: zfs.pools[0]?.name ?? '' })}>+ New schedule diff --git a/web/src/api.ts b/web/src/api.ts index b819c0f..6d20a98 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -268,6 +268,20 @@ export interface ZFSWrites { dirty: number; dirty_max: number; delay_min_pct: number; limit: number; delayed_per_sec: number; over_max_per_sec: number; txg_timeout_sec: number pools: PoolWrites[] } +export interface VerifyPolicy { id: string; name: string; every: 'hourly' | 'daily' | 'weekly' | 'monthly' | 'custom'; at: string; weekday: number; day: number; cron?: string; scope: 'new' | 'all'; enabled: boolean } +export interface VerifyIssue { path: string; kind: 'mismatch' | 'gone' | 'unreadable'; moved: number; size: number; note?: string } +export interface VerifyResult { + id: string; scope: 'new' | 'all'; started: string; finished: string; src: string; dst: string; since: number + entries: number; checked: number; matched: number; mismatched: number; changed: number; gone: number; waiting: number; unreadable: number; unhashed: number + bytes_read: number; issues: VerifyIssue[]; error?: string; cursor: number +} +export interface VerifyProgress { id: string; scope: string; pid: number; started: string; updated: string; total: number; done: number; bytes_read: number; bytes_all: number; current: string } +export interface StaleFile { path: string; size: number; mtime: number } +export interface LandingCheck { + tiers: { src: string; dst: string; ledger: string; cursor: number; stale_count: number; stale_bytes: number; stale: StaleFile[] }[] + stale_after_sec: number; running: VerifyProgress | null; history: VerifyResult[]; cron_file: string + policies: { policy: VerifyPolicy; cron: string; next?: string; last: VerifyResult | null }[] +} export interface ARCView { detail: ARCDetail; live: ARCLive | null; writes: ZFSWrites | null } export interface FileEntry { name: string; type: 'dir' | 'file' | 'link' | 'other'; size: number; mode: string; perm: string; mtime: number; user: string; group: string; target?: string; broken?: boolean; preview?: 'image' | 'video' | 'audio' } @@ -352,6 +366,11 @@ export const api = { removeKVM: (id: string) => req('DELETE', `/api/kvms/${encodeURIComponent(id)}`, undefined, true), zfsOverview: () => req('GET', '/api/zfs'), zfsARC: () => req('GET', '/api/zfs/arc'), + zfsLanding: () => req('GET', '/api/zfs/landing'), + landingVerify: (scope: 'new' | 'all') => req<{ started: string }>('POST', '/api/zfs/landing/verify', { scope }), + saveVerifySchedule: (p: Partial) => req('PUT', '/api/zfs/landing/schedules', p), + deleteVerifySchedule: (id: string) => req('DELETE', `/api/zfs/landing/schedules/${encodeURIComponent(id)}`), + runVerifySchedule: (id: string) => req<{ started: string }>('POST', `/api/zfs/landing/schedules/${encodeURIComponent(id)}/run`), saveZfsSchedule: (p: Partial) => req('PUT', '/api/zfs/schedules', p), deleteZfsSchedule: (id: string) => req('DELETE', `/api/zfs/schedules/${encodeURIComponent(id)}`), runZfsSchedule: (id: string) => req<{ taken: string; pruned: string[] }>('POST', `/api/zfs/schedules/${encodeURIComponent(id)}/run`),