hamlab/internal/storage/landing.go

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
}