hamlab/internal/api/landing.go

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})
}