felhom-agent/internal/hub/loop.go

package hub

import (
	"context"
	"log/slog"
	"time"
)

// interval clamp bounds (locked decision 3).
const (
	MinPollSeconds = 60
	MaxPollSeconds = 3600
)

// reporter and collectorIface are the loop's deps as interfaces (tests inject fakes).
type reporter interface {
	Report(ctx context.Context, r *HostReport) (*ControlEnvelope, error)
}
type collectorIface interface {
	Collect(ctx context.Context) (*HostReport, error)
}

// EnvelopeObserver is notified of the hub's control envelope on every heartbeat (slice 10A).
// The desired-state sync layer (internal/desired) implements it: when DesiredGeneration advances
// past its cache it fetches the full desired-state and updates the engine's provider. Defined
// here (consumer-side) so hub does NOT import the desired/reconcile packages — same seam pattern
// as the collector's StorageObserver. A nil observer (no desired-state wiring) is a clean no-op.
type EnvelopeObserver interface {
	OnEnvelope(ctx context.Context, env *ControlEnvelope)
}

// MultiObserver fans one envelope out to several observers in order (e.g. the desired-state
// syncer + the signed-jobs runner). nil entries are skipped.
func MultiObserver(observers ...EnvelopeObserver) EnvelopeObserver {
	return multiObserver(observers)
}

type multiObserver []EnvelopeObserver

func (m multiObserver) OnEnvelope(ctx context.Context, env *ControlEnvelope) {
	for _, o := range m {
		if o != nil {
			o.OnEnvelope(ctx, env)
		}
	}
}

// Loop is the agent's first daemon run loop: collect a host-report, POST it, adopt
// the hub's cadence, repeat. It is resilient — a collect or report error is logged
// and the loop continues (the data plane is independent of the agent; a hub outage
// must not kill it). There are NO Proxmox mutations here (read-only report), so no
// per-guest work queue yet (that lands with reconcile, slice 4).
type Loop struct {
	collector collectorIface
	client    reporter
	interval  time.Duration
	logger    *slog.Logger
	trigger   <-chan struct{}  // optional: an out-of-band report request (storage watchdog)
	observer  EnvelopeObserver // optional: the slice-10A desired-state sync hook

	// Heartbeat log-pull (v0.83.0): logTailSource yields the debug ring's formatted
	// lines newest-kept within a byte budget (applog.Ring.Lines). logTailPending is
	// armed by an envelope's log_tail_requested and drained onto the NEXT report —
	// the report-channel logtail.go consume-once shape: a failed push leaves the
	// hub's request pending, so the next successful envelope re-arms it (fail-safe
	// retry, no duplicate shipping). Loop state is single-goroutine (cycle only).
	logTailSource  func(maxBytes int) []string
	logTailPending bool
}

// logTailMaxBytes caps the heartbeat log tail (newest lines kept).
const logTailMaxBytes = 128 * 1024

// NewLoop builds the loop. interval is the starting cadence (the hub may override it
// per-cycle via the control envelope).
func NewLoop(collector collectorIface, client reporter, interval time.Duration, logger *slog.Logger) *Loop {
	if logger == nil {
		logger = slog.Default()
	}
	return &Loop{collector: collector, client: client, interval: interval, logger: logger}
}

// SetTrigger wires an out-of-band report channel. A receive on it runs one extra
// collect→report cycle immediately WITHOUT disturbing the regular ticker cadence. Two producers
// fan into this one channel: the storage watchdog (push a disconnect in seconds) and the
// agent-plane poke listener (v0.89.0 immediate-sync). Both debounce, and the channel is cap-1
// non-blocking, so a burst coalesces to at most one pending extra cycle.
func (l *Loop) SetTrigger(ch <-chan struct{}) { l.trigger = ch }

// SetEnvelopeObserver wires the slice-10A desired-state sync hook. It is called once per cycle
// with the hub's control envelope (after the interval is adopted), so the sync layer can fetch
// desired-state when the generation advances. Optional — unset is a clean no-op.
func (l *Loop) SetEnvelopeObserver(o EnvelopeObserver) { l.observer = o }

// SetLogTailSource wires the debug ring for the heartbeat log-pull (v0.83.0).
// Optional — unset means an envelope's log_tail_requested is ignored.
func (l *Loop) SetLogTailSource(src func(maxBytes int) []string) { l.logTailSource = src }

// Run reports immediately, then on each tick, until ctx is cancelled (then nil).
func (l *Loop) Run(ctx context.Context) error {
	interval := l.interval
	interval = l.cycle(ctx, interval) // immediate first report

	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			l.logger.Info("hub: loop shutting down", "reason", ctx.Err())
			return nil
		case <-ticker.C:
			next := l.cycle(ctx, interval)
			if next != interval {
				l.logger.Info("hub: poll interval changed", "from", interval, "to", next)
				interval = next
				ticker.Reset(interval)
			}
		case <-l.trigger:
			// Out-of-band report (storage watchdog OR agent-plane poke). Run a cycle now; keep the
			// regular cadence (do not reset the ticker). The envelope's interval is still adopted
			// if it changed, mirroring the normal path.
			l.logger.Info("hub: out-of-band report triggered (watchdog/poke)")
			next := l.cycle(ctx, interval)
			if next != interval {
				l.logger.Info("hub: poll interval changed", "from", interval, "to", next)
				interval = next
				ticker.Reset(interval)
			}
		}
	}
}

// cycle runs one collect→report→adopt. It never returns an error: failures are
// logged and the current interval is kept, so the loop keeps running.
func (l *Loop) cycle(ctx context.Context, current time.Duration) time.Duration {
	start := time.Now()
	report, err := l.collector.Collect(ctx)
	if err != nil {
		l.logger.Warn("hub: collect failed; skipping this cycle's report", "err", err)
		return current
	}
	// Fulfill a pending log-pull: attach the ring tail to THIS report and clear the
	// local pending flag (consume-once). On a failed push the hub's request is still
	// pending and the next envelope re-arms it — logtail.go's fail-safe retry shape.
	// The explicit nil first makes this robust to a collector reusing its report struct.
	report.LogTail = nil
	if l.logTailPending && l.logTailSource != nil {
		report.LogTail = &LogTail{
			CollectedAt: time.Now().UTC().Format(time.RFC3339),
			Lines:       l.logTailSource(logTailMaxBytes),
		}
	}
	l.logTailPending = false
	env, err := l.client.Report(ctx, report)
	if err != nil {
		l.logger.Warn("hub: report failed; keeping current interval", "err", err)
		return current
	}
	if report.LogTail != nil {
		// Transparency: the pull is visible in the box's own log (and thus in the ring).
		l.logger.Info("operator log pull served", "component", "agent", "lines", len(report.LogTail.Lines))
	}
	l.logger.Debug("hub: report sent",
		"guests", len(report.Guests), "duration_ms", time.Since(start).Milliseconds(),
		"blocked", env.Blocked, "desired_generation", env.DesiredGeneration, "has_signed_ops", env.HasSignedOps)
	if env.LogTailRequested {
		l.logger.Debug("hub: log tail requested — shipping on the next heartbeat")
		l.logTailPending = true
	}

	// Slice 10A: hand the envelope to the desired-state sync hook (fetch desired-state on a
	// generation advance). Done off the report's critical path semantics — a sync/fetch failure
	// is the observer's concern and never affects the heartbeat cadence below.
	if l.observer != nil {
		l.observer.OnEnvelope(ctx, env)
	}

	if env.PollIntervalSeconds == nil {
		return current
	}
	d, clamped := clampInterval(*env.PollIntervalSeconds)
	if clamped {
		l.logger.Warn("hub: poll_interval_seconds out of range; clamped",
			"requested", *env.PollIntervalSeconds, "applied", int(d.Seconds()))
	}
	return d
}

// clampInterval clamps a requested seconds value to [60,3600]; clamped reports
// whether it was out of range.
func clampInterval(sec int) (time.Duration, bool) {
	clamped := false
	if sec < MinPollSeconds {
		sec, clamped = MinPollSeconds, true
	}
	if sec > MaxPollSeconds {
		sec, clamped = MaxPollSeconds, true
	}
	return time.Duration(sec) * time.Second, clamped
}