ZFS tab: landing tier checks — stale files, verify passes against the mover's ledger, verify schedules
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 (its ledger, named in its status file): match, mismatch, edited since, gone, waiting, unreadable, unhashed. 'New moves' checks what moved since the last pass, 'everything' the whole ledger. Passes run as 'hamlab landing-verify' in their own session — the same command a schedule's line in /etc/cron.d/hamlab-landing runs — so a console restart does not cut one short. Stale files: anything waiting on the landing branch over an hour, listed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Mitchell-Johnson
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
f518df239f
commit
829617c932
@@ -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})
|
||||
}
|
||||
@@ -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))
|
||||
|
||||
+5
-1
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user