slice 8B (agent half): /backup/due cadence policy + /backup/status phases (v0.11.0)

internal/localapi: real /backup/due (cadence; due when no successful backup or newest older than backup.backup_cadence_seconds; false in-window after success; failed doesn't count) + /backup/status phases (idle|running|done|failed + job id) + POST /backup single-flight with job id. Drives the controller quiesce loop. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 10:44:50 +02:00
parent e51b3a2f66
commit 33dfd9afb3
6 changed files with 352 additions and 54 deletions
@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"sync"
 	"time"

 	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
@@ -60,7 +61,33 @@ type Options struct {
 	Store      BackupStore
 	Storage    StorageView
 	Tokens     TokenAuthority
-	Logger     *slog.Logger
+	// BackupCadence is the per-guest backup interval driving GET /backup/due (slice 8B). A guest
+	// is "due" when no successful backup is recorded OR the newest one is older than this. 0 → a
+	// safe default (24h). The hub-served policy is slice 10; this is the agent-local cadence.
+	BackupCadence time.Duration
+	Logger        *slog.Logger
+}
+
+// defaultBackupCadence is the fallback /backup/due window when none is configured.
+const defaultBackupCadence = 24 * time.Hour
+
+// Backup phase vocabulary reported by GET /backup/status (slice 8B). The 8B.2 fast-follow adds a
+// `snapshotted` phase (vzdump --mode snapshot) so the controller can unquiesce at snapshot-taken.
+const (
+	PhaseIdle    = "idle"
+	PhaseRunning = "running"
+	PhaseDone    = "done"
+	PhaseFailed  = "failed"
+)
+
+// backupJob is the in-flight/last backup job for one guest (drives /backup/status phases).
+type backupJob struct {
+	JobID      string
+	Phase      string
+	StartedAt  time.Time
+	FinishedAt time.Time
+	Archive    string
+	Error      string
 }

 // Server is the per-guest local API (doc 03 §6). It serves the agent's pinned self-signed leaf
@@ -73,7 +100,12 @@ type Server struct {
 	store   BackupStore
 	storage StorageView
 	tokens  TokenAuthority
+	cadence time.Duration
 	logger  *slog.Logger
+	now     func() time.Time
+
+	jobsMu sync.Mutex
+	jobs   map[int]*backupJob // per-guest backup job state (slice 8B)

 	baseCtx context.Context // for fire-and-forget backups; set in Run
 }
@@ -89,6 +121,10 @@ func NewServer(o Options) (*Server, error) {
 	if o.Logger == nil {
 		o.Logger = slog.Default()
 	}
+	cadence := o.BackupCadence
+	if cadence <= 0 {
+		cadence = defaultBackupCadence
+	}
 	return &Server{
 		addr:    o.ListenAddr,
 		cert:    o.Cert,
@@ -97,7 +133,10 @@ func NewServer(o Options) (*Server, error) {
 		store:   o.Store,
 		storage: o.Storage,
 		tokens:  o.Tokens,
+		cadence: cadence,
 		logger:  o.Logger,
+		now:     func() time.Time { return time.Now().UTC() },
+		jobs:    map[int]*backupJob{},
 	}, nil
 }

@@ -313,6 +352,13 @@ type backupRequest struct {
 	VMID int `json:"vmid"`
 }

+// BackupResponse is POST /backup. The controller polls GET /backup/status on job_id to completion.
+type BackupResponse struct {
+	VMID  int    `json:"vmid"`
+	JobID string `json:"job_id"`
+	Phase string `json:"phase"`
+}
+
 func (s *Server) handleBackup(w http.ResponseWriter, r *http.Request, vmid int) {
 	// Body is optional; if present its vmid must match the token's guest.
 	if r.ContentLength != 0 {
@@ -324,10 +370,24 @@ func (s *Server) handleBackup(w http.ResponseWriter, r *http.Request, vmid int)
 			return
 		}
 	}
-	// Enqueue: a vzdump runs for minutes, so we do not block the request. The backup runs on
-	// the server's base context (cancelled on daemon shutdown) and records into the store; the
-	// controller polls GET /backup/status. This is the crash-consistent path (8A); the
-	// app-consistent quiesce-then-backup loop is 8B.
+	// Single-flight per guest: if a backup is already running for this guest, return that job
+	// (don't start a second concurrent vzdump). The controller polls /backup/status on it.
+	s.jobsMu.Lock()
+	if cur := s.jobs[vmid]; cur != nil && cur.Phase == PhaseRunning {
+		job := *cur
+		s.jobsMu.Unlock()
+		writeStatus(w, http.StatusAccepted, true, BackupResponse{VMID: vmid, JobID: job.JobID, Phase: job.Phase}, "")
+		return
+	}
+	jobID := "backup-" + strconv.Itoa(vmid) + "-" + strconv.FormatInt(s.now().UnixNano(), 10)
+	s.jobs[vmid] = &backupJob{JobID: jobID, Phase: PhaseRunning, StartedAt: s.now()}
+	s.jobsMu.Unlock()
+
+	// Enqueue: a vzdump runs for minutes, so we do not block the request. The backup runs on the
+	// server's base context (cancelled on daemon shutdown), updates the job phase, and records
+	// into the store; the controller polls GET /backup/status. This is the host-side half of the
+	// 8B app-consistent path — the controller quiesces (stops its stacks) BEFORE calling this, so
+	// the vzdump captures a clean-shutdown-consistent state.
 	base := s.baseCtx
 	if base == nil {
 		base = context.Background()
@@ -342,36 +402,94 @@ func (s *Server) handleBackup(w http.ResponseWriter, r *http.Request, vmid int)
 			if b.Error == "" {
 				b.Error = err.Error()
 			}
-			s.logger.Error("local-api: enqueued backup failed", "vmid", vmid, "err", err)
+			s.logger.Error("local-api: backup job failed", "vmid", vmid, "job", jobID, "err", err)
 		} else {
-			s.logger.Info("local-api: enqueued backup complete", "vmid", vmid, "archive", b.Archive)
+			s.logger.Info("local-api: backup job complete", "vmid", vmid, "job", jobID, "archive", b.Archive)
 		}
 		s.store.RecordBackup(b)
+		s.finishJob(vmid, jobID, b)
 	}()
-	writeStatus(w, http.StatusAccepted, true, map[string]any{"vmid": vmid, "enqueued": true}, "")
+	writeStatus(w, http.StatusAccepted, true, BackupResponse{VMID: vmid, JobID: jobID, Phase: PhaseRunning}, "")
 }

-// BackupDueResponse is GET /backup/due. Thin in 8A: a guest with no successful backup recorded
-// is "due"; otherwise not. Policy-scheduled cadence (hub manifest) lands in slice 10, and the
-// quiesce-on-due consumer is 8B — both noted in the response.
+// finishJob transitions the guest's job to done/failed (only if it is still the current job — a
+// later job started after a single-flight gap must not be overwritten by an older one's result).
+func (s *Server) finishJob(vmid int, jobID string, b hub.Backup) {
+	s.jobsMu.Lock()
+	defer s.jobsMu.Unlock()
+	cur := s.jobs[vmid]
+	if cur == nil || cur.JobID != jobID {
+		return
+	}
+	cur.FinishedAt = s.now()
+	if b.Success {
+		cur.Phase = PhaseDone
+		cur.Archive = b.Archive
+	} else {
+		cur.Phase = PhaseFailed
+		cur.Error = b.Error
+	}
+}
+
+// jobSnapshot returns a copy of the guest's current job (ok=false if none).
+func (s *Server) jobSnapshot(vmid int) (backupJob, bool) {
+	s.jobsMu.Lock()
+	defer s.jobsMu.Unlock()
+	if j := s.jobs[vmid]; j != nil {
+		return *j, true
+	}
+	return backupJob{}, false
+}
+
+// BackupDueResponse is GET /backup/due (slice 8B). A guest is due when no successful backup is
+// recorded OR the newest successful one is older than the agent-local cadence. A successful
+// POST /backup flips this to false for the window, so the controller won't re-quiesce in a loop.
+// The hub-served policy is slice 10.
 type BackupDueResponse struct {
-	VMID   int    `json:"vmid"`
-	Due    bool   `json:"due"`
-	Reason string `json:"reason"`
+	VMID    int    `json:"vmid"`
+	Due     bool   `json:"due"`
+	Reason  string `json:"reason"`
+	AgeSecs *int64 `json:"age_seconds,omitempty"` // age of the newest successful backup; null if none
 }

 func (s *Server) handleBackupDue(w http.ResponseWriter, r *http.Request, vmid int) {
-	latest := s.latestBackupFor(r.Context(), vmid)
-	if latest == nil || !latest.Success {
+	latest := s.latestSuccessfulBackupFor(r.Context(), vmid)
+	if latest == nil {
 		writeOK(w, BackupDueResponse{VMID: vmid, Due: true, Reason: "no successful backup recorded yet"})
 		return
 	}
-	writeOK(w, BackupDueResponse{VMID: vmid, Due: false, Reason: "policy-scheduled cadence lands in slice 10"})
+	age, ok := backupAge(latest.StartedAt, s.now())
+	if !ok {
+		// Unparseable timestamp: fail safe toward "due" so a backup still happens.
+		writeOK(w, BackupDueResponse{VMID: vmid, Due: true, Reason: "last backup time unparseable — treating as due"})
+		return
+	}
+	ageSecs := int64(age.Seconds())
+	if age >= s.cadence {
+		writeOK(w, BackupDueResponse{VMID: vmid, Due: true, Reason: "older than cadence", AgeSecs: &ageSecs})
+		return
+	}
+	writeOK(w, BackupDueResponse{VMID: vmid, Due: false, Reason: "within cadence window", AgeSecs: &ageSecs})
+}
+
+// BackupStatusResponse is GET /backup/status (slice 8B): the current/last job phase + the latest
+// recorded backup. Phase is idle when no job has run this process lifetime.
+type BackupStatusResponse struct {
+	VMID    int         `json:"vmid"`
+	Phase   string      `json:"phase"`             // idle | running | done | failed
+	JobID   string      `json:"job_id,omitempty"`
+	Error   string      `json:"error,omitempty"`
+	Backup  *hub.Backup `json:"backup,omitempty"`  // latest recorded backup for this guest
 }

 func (s *Server) handleBackupStatus(w http.ResponseWriter, r *http.Request, vmid int) {
-	latest := s.latestBackupFor(r.Context(), vmid)
-	writeOK(w, map[string]any{"vmid": vmid, "backup": latest})
+	resp := BackupStatusResponse{VMID: vmid, Phase: PhaseIdle, Backup: s.latestBackupFor(r.Context(), vmid)}
+	if job, ok := s.jobSnapshot(vmid); ok {
+		resp.Phase = job.Phase
+		resp.JobID = job.JobID
+		resp.Error = job.Error
+	}
+	writeOK(w, resp)
 }

 func (s *Server) handleRestoreTestStatus(w http.ResponseWriter, r *http.Request, vmid int) {
@@ -389,9 +507,19 @@ func (s *Server) handleRestoreTestStatus(w http.ResponseWriter, r *http.Request,

 // latestBackupFor returns this guest's most recent backup from the store (nil if none).
 func (s *Server) latestBackupFor(ctx context.Context, vmid int) *hub.Backup {
+	return s.pickLatestBackup(ctx, vmid, false)
+}
+
+// latestSuccessfulBackupFor returns this guest's most recent SUCCESSFUL backup (nil if none) —
+// the basis for /backup/due (a failed backup must not satisfy the cadence).
+func (s *Server) latestSuccessfulBackupFor(ctx context.Context, vmid int) *hub.Backup {
+	return s.pickLatestBackup(ctx, vmid, true)
+}
+
+func (s *Server) pickLatestBackup(ctx context.Context, vmid int, successOnly bool) *hub.Backup {
 	var latest *hub.Backup
 	for _, b := range s.store.Backups(ctx) {
-		if b.VMID != vmid {
+		if b.VMID != vmid || (successOnly && !b.Success) {
 			continue
 		}
 		bb := b
@@ -402,6 +530,15 @@ func (s *Server) latestBackupFor(ctx context.Context, vmid int) *hub.Backup {
 	return latest
 }

+// backupAge parses an RFC3339 backup start time and returns its age relative to now.
+func backupAge(startedAt string, now time.Time) (time.Duration, bool) {
+	t, err := time.Parse(time.RFC3339, startedAt)
+	if err != nil {
+		return 0, false
+	}
+	return now.Sub(t), true
+}
+
 // classByStorage builds a storage-id → class(fast|slow|"") map from the host storage view. A
 // view error is logged and yields an empty map (class falls back to "" — a hint, not load-bearing).
 func (s *Server) classByStorage(ctx context.Context) map[string]string {