Release v0.30.0-beta7

cmd/proxsave/backup_stream.go

@@ -27,6 +27,40 @@ var backupStreamSteps = runBackupModeSteps
 // session so the emitted lines and the outcome can be asserted.
 var backupAdoptSession = adoptDashboardSession
 
+// dashboardStreamBacklogMaxLines bounds the pre-viewport backlog. A run's early
+// phase (banner -> environment -> security preflight, before the backup steps)
+// logs on the order of a few dozen lines at INFO and a couple hundred at DEBUG, so
+// this never drops in practice; it only backstops a pathological producer.
+const dashboardStreamBacklogMaxLines = 4096
+
+// dashboardStreamBacklog holds the run logger's colored line stream captured
+// BEFORE the graphical viewport exists (wired by initializeRunLogger on a
+// dashboard handoff). runBackupStreamed drains it into the viewport, then detaches
+// the mirror so the live stream is not double-captured. Single-use per process
+// (proxsave runs once); nil when there is no handoff (CLI/cron/daemon).
+var dashboardStreamBacklog *logging.LineBacklog
+
+// replayDashboardStreamBacklog flushes the pre-viewport backlog into the streamed
+// panel, in order, BEFORE the live capture is installed, so the viewport shows the
+// SAME complete flow the on-disk log has instead of starting mid-run. It first
+// detaches the logger mirror, THEN emits each retained line; the caller installs
+// captureRunOutput only after this returns, so no line can ever reach both the
+// backlog and the live pipe. A nil backlog (no handoff) is a no-op. Called once.
+func replayDashboardStreamBacklog(emit func(line string)) {
+	backlog := dashboardStreamBacklog
+	dashboardStreamBacklog = nil
+	if backlog == nil {
+		return
+	}
+	// Detach the mirror while the console is still muted and before the live pipe
+	// exists: from here the run's remaining lines reach the panel only through the
+	// capture the caller installs next, so nothing is double-captured.
+	logging.GetDefaultLogger().SetMirror(nil)
+	for _, line := range backlog.Lines() {
+		emit(line)
+	}
+}
+
 // captureRunOutput routes BOTH the loggers (default + colored bootstrap mirror)
 // AND raw os.Stdout through a SINGLE pipe into emit, so everything a run prints -
 // colored logger lines AND the raw fmt.Println blank spacers between sections -
@@ -115,6 +149,13 @@ func runBackupStreamed(opts backupModeOptions) backupModeResult {
 	var res backupModeResult
 	streamErr := components.RunStreamTask(opts.ctx, session, "Running backup",
 		func(taskCtx context.Context, emit func(line string)) (string, error) {
+			// Replay everything logged BEFORE this viewport existed (banner,
+			// environment, identity, security preflight) so the panel shows the
+			// same complete run the on-disk log has. Done FIRST - it detaches the
+			// mirror before captureRunOutput installs the live pipe, so no line can
+			// ever reach both the backlog and the live capture (no double).
+			replayDashboardStreamBacklog(emit)
+
 			// Route the default + COLORED bootstrap-mirror loggers AND raw os.Stdout
 			// (the fmt.Println section spacers the CLI prints) through one pipe into
 			// the panel; restored on return/panic. So the panel shows the SAME lines -

cmd/proxsave/backup_stream_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"io"
 	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -516,6 +517,87 @@ func TestRunBackupStreamedHandsOffInStream(t *testing.T) {
 	_ = pumpEnterBackup(t, session, resCh)
 }
 
+// TestRunBackupStreamedReplaysPreStreamBacklog is the guard for the fix: the
+// streamed viewport must show the lines logged BEFORE it existed (banner ->
+// environment -> preflight), not start mid-run at "Initializing backup
+// orchestrator". It reproduces the pre-viewport phase the way initializeRunLogger
+// wires it on a dashboard handoff - the run logger mirrors its full colored stream
+// into a backlog while the console is muted - including a RAW banner line flushed
+// via AppendRaw (which the console path deliberately skips, so only the mirror
+// carries it). runBackupStreamed must replay that backlog into the panel ahead of
+// the live step line.
+func TestRunBackupStreamedReplaysPreStreamBacklog(t *testing.T) {
+	prevLogger := logging.GetDefaultLogger()
+	lg := logging.New(types.LogLevelInfo, false)
+	lg.SetOutput(io.Discard) // muted console, as during a dashboard handoff
+	logging.SetDefaultLogger(lg)
+	t.Cleanup(func() { logging.SetDefaultLogger(prevLogger) })
+
+	// Wire the mirror + backlog exactly like initializeRunLogger, then log the
+	// early lines: a raw banner (file/mirror only, never on the console path) and a
+	// normal INFO line. An open log file lets AppendRaw exercise its file branch too.
+	backlog := logging.NewLineBacklog(1024)
+	lg.SetMirror(backlog)
+	if err := lg.OpenLogFile(filepath.Join(t.TempDir(), "run.log")); err != nil {
+		t.Fatalf("open log file: %v", err)
+	}
+	lg.AppendRaw("===== ProxSaveBanner =====")
+	logging.Info("Environment: dual pve=9.1.9")
+	dashboardStreamBacklog = backlog
+	t.Cleanup(func() { dashboardStreamBacklog = nil })
+
+	prevSteps := backupStreamSteps
+	backupStreamSteps = func(opts backupModeOptions) backupModeResult {
+		logging.Info("Initializing backup orchestrator")
+		return backupModeResult{
+			exitCode:     types.ExitSuccess.Int(),
+			supportStats: &orchestrator.BackupStats{FilesCollected: 1, LocalStatus: "ok"},
+		}
+	}
+	t.Cleanup(func() { backupStreamSteps = prevSteps })
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var buf shell.SyncBuffer
+	session := shell.StartForTestWithOutput(ctx, shell.Config{AppName: "ProxSave", Subtitle: "Backup"}, &buf)
+
+	bootstrap := logging.NewBootstrapLogger()
+	stashDashboardSession(session, bootstrap)
+	t.Cleanup(releaseDashboardLeftovers)
+
+	resCh := make(chan backupModeResult, 1)
+	go func() {
+		resCh <- runBackupStreamed(backupModeOptions{ctx: ctx, bootstrap: bootstrap, cfg: &config.Config{}})
+	}()
+
+	// The pre-viewport backlog (banner via AppendRaw + the environment line) must
+	// stream into the panel, followed by the live step line.
+	waitFor(t, &buf, "ProxSaveBanner")
+	waitFor(t, &buf, "Environment: dual")
+	waitFor(t, &buf, "Initializing backup orchestrator")
+
+	// Presence is not enough: the whole point is order. The replayed backlog must
+	// precede the live step line, exactly as the on-disk log records it
+	// (banner -> environment -> orchestrator init). A regression that streamed the
+	// step line first would still satisfy the three waitFor calls above.
+	frame := ansi.Strip(buf.String())
+	iBanner := strings.Index(frame, "ProxSaveBanner")
+	iEnv := strings.Index(frame, "Environment: dual")
+	iInit := strings.Index(frame, "Initializing backup orchestrator")
+	if iBanner < 0 || iBanner >= iEnv || iEnv >= iInit {
+		t.Fatalf("stream out of order: banner=%d environment=%d init=%d (want banner < environment < init)", iBanner, iEnv, iInit)
+	}
+
+	_ = pumpEnterBackup(t, session, resCh)
+
+	// The mirror is detached once the backlog is replayed, so a line logged after
+	// the run does not keep feeding the (now stale) backlog.
+	if got := logging.GetDefaultLogger(); got != nil {
+		got.SetMirror(nil) // idempotent; asserts the symbol exists / no panic
+	}
+}
+
 // pumpEnterBackup sends Enter until runBackupStreamed returns its result.
 func pumpEnterBackup(t *testing.T, s *shell.Session, done <-chan backupModeResult) backupModeResult {
 	t.Helper()

cmd/proxsave/daemon.go

@@ -5,7 +5,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"hash/fnv"
 	"io"
+	"math"
 	"os"
 	"os/exec"
 	"os/signal"
@@ -77,11 +79,11 @@ type daemon struct {
 	configPath string
 	now        func() time.Time
 
-	mu            sync.Mutex
-	reporter      backupReporter
-	fetchWarned   bool      // centralized fetch already warned once (throttle recurring WARN)
-	updateWarned  bool      // an update is already known available (WARN once per transition)
-	provisionLast time.Time // last relay-secret self-heal attempt (throttle); guarded by mu
+	mu               sync.Mutex
+	reporter         backupReporter
+	fetchWarned      bool      // centralized fetch already warned once (throttle recurring WARN)
+	updateWarned     bool      // an update is already known available (WARN once per transition)
+	provisionRetryAt time.Time // next relay-secret self-heal attempt; guarded by mu
 	// newBackupCmd builds the child backup command; overridable in tests.
 	newBackupCmd func(ctx context.Context) *exec.Cmd
 
@@ -628,22 +630,25 @@ func (d *daemon) buildReporter(ctx context.Context) *health.Reporter {
 	// centralized
 	alive, backup, checks, secretUsed, err := d.fetchCentralized(ctx)
 	if err != nil {
-		// A DEFINITIVE auth rejection (ErrHCAuth) means the on-disk relay secret no longer
-		// matches the server's stored hash (a server DB restore/rollback, or a lost
-		// double-issuance race). Clear it so the next throttled maybeProvisionRelaySecret
-		// mints a fresh one - restoring the Telegram path's self-heal. ONLY ErrHCAuth: a
-		// transient / unreachable / not-ready / unknown error must NOT churn a
-		// possibly-good secret (a working confirmed secret never returns 401/403). The clear
-		// is value-guarded under LockNotifySecret against the EXACT secret fetchCentralized
-		// used (secretUsed), so a concurrent hook that persisted+confirmed a fresh secret in
-		// the meantime is never clobbered (which would strand the host).
-		if errors.Is(err, health.ErrHCAuth) {
+		// A DEFINITIVE rejection means the on-disk relay secret is no longer usable:
+		//   ErrHCAuth   - the secret no longer matches the server's stored hash (a
+		//                 server DB restore/rollback, or a lost double-issuance race);
+		//   ErrHCParked - the server purged this host's row (its unused account was
+		//                 parked, design 11.2), so the token authenticates nothing.
+		// Both clear the secret so the next throttled maybeProvisionRelaySecret mints a
+		// fresh one (a parked host is re-admitted as a recreation) - restoring the
+		// self-heal. ONLY these two: a transient / unreachable / not-ready / unknown
+		// error must NOT churn a possibly-good secret. The clear is value-guarded under
+		// LockNotifySecret against the EXACT secret fetchCentralized used (secretUsed),
+		// so a concurrent hook that persisted+confirmed a fresh secret is never
+		// clobbered (which would strand the host).
+		if errors.Is(err, health.ErrHCAuth) || errors.Is(err, health.ErrHCParked) {
 			if cleared, rmErr := identity.RemoveNotifySecretIfMatches(d.cfg.BaseDir, secretUsed); rmErr != nil {
 				logging.Debug("daemon: clear rejected relay secret failed: %v", rmErr)
 			} else if cleared {
-				logging.Debug("daemon: relay secret rejected by server (auth); cleared for re-provisioning")
+				logging.Debug("daemon: relay secret rejected by server (auth/parked); cleared for re-provisioning")
 			} else {
-				logging.Debug("daemon: relay secret rejected by server (auth) but on-disk secret changed concurrently; keeping it")
+				logging.Debug("daemon: relay secret rejected by server (auth/parked) but on-disk secret changed concurrently; keeping it")
 			}
 		}
 		// The heartbeat loop retries this every interval; warn ONCE (so the
@@ -710,61 +715,112 @@ func (d *daemon) fetchCentralized(ctx context.Context) (alive, backup string, ch
 	return cfg.AliveURL, cfg.BackupURL, cfg.Checks, secret, nil
 }
 
-// daemonProvisionRetryInterval throttles the daemon's relay-secret self-heal so a persistent
-// provisioning failure (server down, or the host not yet known to the server) does not hit the
-// server on every heartbeat interval.
-const daemonProvisionRetryInterval = 15 * time.Minute
+const (
+	// daemonProvisionRetryInterval is the local retry floor: a persistent failure
+	// must not hit the relay on every heartbeat interval.
+	daemonProvisionRetryInterval = 15 * time.Minute
+	// daemonProvisionMaxRetryJitter spreads clients released from the same global
+	// admission window without materially extending the server's Retry-After.
+	daemonProvisionMaxRetryJitter = 5 * time.Minute
+)
+
+// daemonProvisionRetryJitter is deterministic per server_id, so no shared random
+// source or persisted state is needed. It uses up to 10% of Retry-After, capped at
+// five minutes. A stable spread is sufficient to avoid a synchronized retry herd.
+func daemonProvisionRetryJitter(serverID string, retryAfter time.Duration) time.Duration {
+	window := retryAfter / 10
+	if window > daemonProvisionMaxRetryJitter {
+		window = daemonProvisionMaxRetryJitter
+	}
+	if window <= 0 {
+		return 0
+	}
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(serverID))
+	// Mask to 63 bits so the sum is a valid non-negative int64 (the conversion cannot
+	// overflow), then reduce it modulo the window in signed space. window is in
+	// (0, 5min], so int64(window)+1 is a small positive divisor and the remainder is a
+	// bounded Duration.
+	return time.Duration(int64(h.Sum64()&math.MaxInt64) % (int64(window) + 1))
+}
 
 // provisionRelaySecretFn is the relay-secret provisioner seam (stubbed in tests so the
 // self-heal logic is exercised without touching the network).
 var provisionRelaySecretFn = notify.ProvisionRelaySecret
 
-// provisionRelaySecretBestEffort attempts one relay-secret provisioning for a centralized
-// healthcheck host that has none on disk, so centralized monitoring works WITHOUT Telegram
-// pairing. Best-effort and non-blocking: it returns the relay secret now on disk, or "" when
-// provisioning is not applicable (self mode / disabled / no ServerID) or the attempt failed.
-// It NEVER returns/propagates an error. When a secret is already present it returns it
-// unchanged (never overwrites a possibly-confirmed secret). Callers gate frequency: the
-// daemon throttles via provisionLast; the one-shot daemon-setup path runs it once.
-func provisionRelaySecretBestEffort(ctx context.Context, cfg *config.Config, logger *logging.Logger) string {
+// provisionRelaySecretAttempt performs the best-effort attempt and also returns
+// server-directed backpressure. It never propagates an error: Retry-After is the
+// only semantic detail the daemon needs from a handled 429.
+func provisionRelaySecretAttempt(
+	ctx context.Context, cfg *config.Config, logger *logging.Logger,
+) (string, time.Duration) {
 	if cfg == nil || !cfg.HealthcheckEnabled || cfg.HealthcheckMode != config.HealthcheckModeCentralized {
-		return ""
+		return "", 0
 	}
 	baseDir := strings.TrimSpace(cfg.BaseDir)
 	if baseDir == "" {
-		return ""
+		return "", 0
 	}
 	if s, _ := identity.LoadNotifySecret(baseDir); strings.TrimSpace(s) != "" {
-		return strings.TrimSpace(s) // already provisioned; do not churn
+		return strings.TrimSpace(s), 0 // already provisioned; do not churn
 	}
 	serverID := strings.TrimSpace(cfg.ServerID)
 	if serverID == "" {
-		return ""
+		return "", 0
 	}
 	if _, err := provisionRelaySecretFn(ctx, cfg.ServerAPIHost, serverID, baseDir, logger); err != nil {
+		var limited *notify.RelayProvisionRateLimitError
+		if errors.As(err, &limited) {
+			logging.Debug(
+				"daemon: relay-secret provisioning rate limited (server backoff honored)")
+			return "", limited.RetryAfter
+		}
 		logging.Debug("daemon: relay-secret provisioning attempt failed (will retry later): %v", err)
-		return ""
+		return "", 0
 	}
 	// Reload regardless of the provisioned flag: ProvisionRelaySecret returns false when it
 	// adopts a secret a concurrent provisioner persisted under the cross-process lock, so a
 	// usable secret may be on disk even then; LoadNotifySecret yields "" when there is
 	// genuinely none, degrading gracefully.
 	s, _ := identity.LoadNotifySecret(baseDir)
-	return strings.TrimSpace(s)
+	return strings.TrimSpace(s), 0
+}
+
+// provisionRelaySecretBestEffort keeps the one-shot setup callers' established
+// string-only contract. They run once and therefore do not need Retry-After.
+func provisionRelaySecretBestEffort(
+	ctx context.Context, cfg *config.Config, logger *logging.Logger,
+) string {
+	secret, _ := provisionRelaySecretAttempt(ctx, cfg, logger)
+	return secret
 }
 
 // maybeProvisionRelaySecret is the daemon's throttled relay-secret self-heal (hook b). It
 // returns a freshly persisted secret, or "" when throttled / not applicable / failed. The
-// throttle is recorded even on failure so a down server is not hit every heartbeat.
+// local retry floor is installed BEFORE the network call so concurrent heartbeat
+// paths cannot duplicate an attempt. A longer server Retry-After extends that
+// deadline and adds bounded deterministic jitter.
 func (d *daemon) maybeProvisionRelaySecret(ctx context.Context) string {
+	now := d.now()
 	d.mu.Lock()
-	if !d.provisionLast.IsZero() && d.now().Sub(d.provisionLast) < daemonProvisionRetryInterval {
+	if !d.provisionRetryAt.IsZero() && now.Before(d.provisionRetryAt) {
 		d.mu.Unlock()
 		return ""
 	}
-	d.provisionLast = d.now()
+	d.provisionRetryAt = now.Add(daemonProvisionRetryInterval)
 	d.mu.Unlock()
-	return provisionRelaySecretBestEffort(ctx, d.cfg, d.logger)
+
+	secret, retryAfter := provisionRelaySecretAttempt(ctx, d.cfg, d.logger)
+	if retryAfter > daemonProvisionRetryInterval {
+		retryAt := d.now().Add(
+			retryAfter + daemonProvisionRetryJitter(d.cfg.ServerID, retryAfter))
+		d.mu.Lock()
+		if retryAt.After(d.provisionRetryAt) {
+			d.provisionRetryAt = retryAt
+		}
+		d.mu.Unlock()
+	}
+	return secret
 }
 
 // provisionRelaySecretOnDaemonSetup is the one-shot relay-secret self-heal (hook a) run during