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
}

// 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 — used by
// the storage watchdog to push a disconnect to the hub in seconds. The watchdog debounces,
// so this fires at most once per debounce window.
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 }

// 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). 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 (storage watchdog)")
			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 {
	report, err := l.collector.Collect(ctx)
	if err != nil {
		l.logger.Warn("hub: collect failed; skipping this cycle's report", "err", err)
		return current
	}
	env, err := l.client.Report(ctx, report)
	if err != nil {
		l.logger.Warn("hub: report failed; keeping current interval", "err", err)
		return current
	}
	l.logger.Debug("hub: report sent",
		"guests", len(report.Guests),
		"blocked", env.Blocked, "desired_generation", env.DesiredGeneration, "has_signed_ops", env.HasSignedOps)

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