slice 8B.2 (agent): emit snapshotted phase at storage-snapshot moment (v0.13.0)

BackupRunner.BackupWithSnapshotHook tails the task log for the 'create storage snapshot' marker (snapshot mode only) and fires onSnapshot once; localapi flips /backup/status to 'snapshotted' before 'done' so the controller resumes early. Phase 0 validated on PVE 9.2.2: marker confirmed, downtime ~24s->~1s (934MB). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 14:54:18 +02:00
parent fe7b3c4ab7
commit 570410cd1a
8 changed files with 248 additions and 18 deletions
@@ -25,7 +25,8 @@ type fakeBackupAPI struct {
 	content    []proxmox.StorageContent
 	contentErr error
 	vzdumps    []proxmox.VzdumpOptions
-	logLines   []string // returned by TaskLogTail (e.g. "INFO: backup mode: stop")
+	logLines   []string      // returned by TaskLogTail (e.g. "INFO: backup mode: stop")
+	waitGate   chan struct{} // if non-nil, WaitTask blocks until closed (8B.2 watcher timing)
 }

 func (f *fakeBackupAPI) Vzdump(_ context.Context, o proxmox.VzdumpOptions) (string, error) {
@@ -33,6 +34,9 @@ func (f *fakeBackupAPI) Vzdump(_ context.Context, o proxmox.VzdumpOptions) (stri
 	return f.vzdumpUPID, f.vzdumpErr
 }
 func (f *fakeBackupAPI) WaitTask(_ context.Context, _ string, _ proxmox.WaitOptions) (proxmox.TaskStatus, error) {
+	if f.waitGate != nil {
+		<-f.waitGate
+	}
 	return proxmox.TaskStatus{Status: "stopped", ExitStatus: "OK"}, f.waitErr
 }
 func (f *fakeBackupAPI) GuestConfig(_ context.Context, _ int) (proxmox.GuestConfig, error) {
@@ -48,10 +48,33 @@ func NewBackupRunner(api BackupAPI, target string, mode proxmox.BackupMode, note
 	return &BackupRunner{api: api, target: target, mode: mode, notes: notes, logger: logger, now: func() time.Time { return time.Now().UTC() }}
 }

-// Backup runs one vzdump of vmid to the local target and returns the report record. A
-// failure is returned BOTH as an error and as a Backup{Success:false,...} so the caller can
-// record the failed attempt.
+// snapshotMarker is the vzdump task-log line that signals the storage snapshot has been created
+// and the backup is now reading from it — the point after which resuming the guest's app cannot
+// affect the backup (slice 8B.2; validated on PVE 9.2.2: `INFO: create storage snapshot 'vzdump'`).
+// It only appears in snapshot mode (stop mode takes no storage snapshot), so its presence ⟹
+// snapshot mode — the basis for the controller's early resume.
+const snapshotMarker = "create storage snapshot"
+
+// snapshotWatchInterval is how often watchForSnapshot polls the task log. A package var so tests
+// can shrink it (production: poll once a second — the marker appears in the first ~1s, §0).
+var snapshotWatchInterval = time.Second
+
+// Backup runs one vzdump of vmid to the local target and returns the report record.
 func (r *BackupRunner) Backup(ctx context.Context, vmid int) (hub.Backup, error) {
+	return r.backup(ctx, vmid, nil)
+}
+
+// BackupWithSnapshotHook is Backup plus an onSnapshot callback invoked ONCE, mid-backup, when the
+// storage snapshot has been taken (snapshot mode only) — the 8B.2 early-resume signal. In
+// stop/downgraded mode the marker never appears, so onSnapshot is never called (the caller then
+// resumes at completion). onSnapshot must be cheap + non-blocking (it runs on a watcher goroutine).
+func (r *BackupRunner) BackupWithSnapshotHook(ctx context.Context, vmid int, onSnapshot func()) (hub.Backup, error) {
+	return r.backup(ctx, vmid, onSnapshot)
+}
+
+// backup is the shared body. A failure is returned BOTH as an error and as a
+// Backup{Success:false,...} so the caller can record the failed attempt.
+func (r *BackupRunner) backup(ctx context.Context, vmid int, onSnapshot func()) (hub.Backup, error) {
 	start := r.now()
 	rec := hub.Backup{
 		TargetID:        r.target,
@@ -83,6 +106,13 @@ func (r *BackupRunner) Backup(ctx context.Context, vmid int) (hub.Backup, error)
 		return rec, fmt.Errorf("backup: vzdump vmid %d: %w", vmid, err)
 	}
 	if upid != "" {
+		// 8B.2: while the backup runs, watch the task log for the storage-snapshot marker and
+		// fire onSnapshot once (snapshot mode only) so the controller can resume its app early.
+		if onSnapshot != nil {
+			watchCtx, stopWatch := context.WithCancel(ctx)
+			defer stopWatch()
+			go r.watchForSnapshot(watchCtx, upid, onSnapshot)
+		}
 		if _, err := r.api.WaitTask(ctx, upid, proxmox.WaitOptions{Timeout: 30 * time.Minute}); err != nil {
 			rec.Error = err.Error()
 			rec.DurationSeconds = time.Since(start).Seconds()
@@ -113,6 +143,36 @@ func (r *BackupRunner) Backup(ctx context.Context, vmid int) (hub.Backup, error)
 	return rec, nil
 }

+// watchForSnapshot polls the running backup's task log until it sees the storage-snapshot marker
+// (→ onSnapshot once) or the requested mode is reported as `stop` (→ downgraded; the marker will
+// never come, so stop watching) or ctx is cancelled (backup finished). Best-effort: a log-read
+// error is retried on the next tick; onSnapshot fires at most once.
+func (r *BackupRunner) watchForSnapshot(ctx context.Context, upid string, onSnapshot func()) {
+	ticker := time.NewTicker(snapshotWatchInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			lines, err := r.api.TaskLogTail(ctx, upid, 200)
+			if err != nil {
+				continue
+			}
+			// A stop-mode (or downgraded) backup never creates a storage snapshot → never resume early.
+			if m := parseBackupMode(lines); m != "" && m != string(proxmox.ModeSnapshot) {
+				return
+			}
+			for _, ln := range lines {
+				if strings.Contains(ln, snapshotMarker) {
+					onSnapshot()
+					return
+				}
+			}
+		}
+	}
+}
+
 // PickRestoreCandidate returns the newest backup archive on the target (any guest), or ""
 // when there is none — the restore-test then no-ops cleanly.
 func (r *BackupRunner) PickRestoreCandidate(ctx context.Context) (string, error) {
@@ -0,0 +1,72 @@
+package backup
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/proxmox"
+)
+
+// 8B.2: in snapshot mode, the runner fires onSnapshot when the storage-snapshot marker appears in
+// the task log — mid-backup, before completion.
+func TestBackupWithSnapshotHook_FiresOnMarker(t *testing.T) {
+	old := snapshotWatchInterval
+	snapshotWatchInterval = 2 * time.Millisecond
+	defer func() { snapshotWatchInterval = old }()
+
+	api := &fakeBackupAPI{
+		vzdumpUPID: "UPID:backup",
+		waitGate:   make(chan struct{}), // hold the backup open so the watcher gets to poll
+		logLines:   []string{"INFO: backup mode: snapshot", "INFO: create storage snapshot 'vzdump'"},
+		content:    []proxmox.StorageContent{{VolID: "local:backup/vzdump-lxc-9001-x", Content: "backup", VMID: 9001, Size: 100, CTime: 1}},
+	}
+	r := NewBackupRunner(api, "local", "", "", quiet())
+
+	var fired int32
+	done := make(chan struct{})
+	go func() {
+		_, _ = r.BackupWithSnapshotHook(context.Background(), 9001, func() { atomic.StoreInt32(&fired, 1) })
+		close(done)
+	}()
+
+	// the watcher should fire onSnapshot well before we release the backup
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) && atomic.LoadInt32(&fired) == 0 {
+		time.Sleep(2 * time.Millisecond)
+	}
+	if atomic.LoadInt32(&fired) != 1 {
+		t.Fatal("onSnapshot did not fire on the storage-snapshot marker")
+	}
+	close(api.waitGate) // let the backup complete
+	<-done
+}
+
+// Stop/downgraded mode → no storage-snapshot marker → onSnapshot never fires (the 8B.2 fallback).
+func TestBackupWithSnapshotHook_StopMode_NeverFires(t *testing.T) {
+	old := snapshotWatchInterval
+	snapshotWatchInterval = 2 * time.Millisecond
+	defer func() { snapshotWatchInterval = old }()
+
+	api := &fakeBackupAPI{
+		vzdumpUPID: "UPID:backup",
+		waitGate:   make(chan struct{}),
+		logLines:   []string{"INFO: backup mode: stop"}, // downgraded; no snapshot marker
+		content:    []proxmox.StorageContent{{VolID: "local:backup/vzdump-lxc-9001-x", Content: "backup", VMID: 9001, Size: 100, CTime: 1}},
+	}
+	r := NewBackupRunner(api, "local", "", "", quiet())
+
+	var fired int32
+	done := make(chan struct{})
+	go func() {
+		_, _ = r.BackupWithSnapshotHook(context.Background(), 9001, func() { atomic.StoreInt32(&fired, 1) })
+		close(done)
+	}()
+	time.Sleep(40 * time.Millisecond) // the watcher polls several times + sees stop mode → returns
+	if atomic.LoadInt32(&fired) != 0 {
+		t.Fatal("onSnapshot fired in stop mode (must not)")
+	}
+	close(api.waitGate)
+	<-done
+}