felhom-agent/internal/hub/loop_test.go

package hub

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func intPtr(i int) *int { return &i }

type fakeCollector struct {
	report *HostReport
	err    error
	n      *int32
}

func (c *fakeCollector) Collect(ctx context.Context) (*HostReport, error) {
	atomic.AddInt32(c.n, 1)
	return c.report, c.err
}

type fakeReporter struct {
	env    *ControlEnvelope
	errSeq []error // per-call error (nil = ok); calls past the slice are ok
	n      *int32
}

func (r *fakeReporter) Report(ctx context.Context, _ *HostReport) (*ControlEnvelope, error) {
	i := int(atomic.AddInt32(r.n, 1) - 1)
	if i < len(r.errSeq) && r.errSeq[i] != nil {
		return nil, r.errSeq[i]
	}
	return r.env, nil
}

// recordingObserver records the envelopes the loop hands it (slice 10A EnvelopeObserver seam).
type recordingObserver struct{ envs []*ControlEnvelope }

func (o *recordingObserver) OnEnvelope(_ context.Context, e *ControlEnvelope) {
	o.envs = append(o.envs, e)
}

// The loop notifies the EnvelopeObserver once per successful cycle (with the envelope) AND still
// adopts PollIntervalSeconds — the two are independent.
func TestLoop_CycleNotifiesObserverAndAdoptsInterval(t *testing.T) {
	var cn, rn int32
	env := &ControlEnvelope{DesiredGeneration: 3, HasSignedOps: true, PollIntervalSeconds: intPtr(120)}
	loop := NewLoop(
		&fakeCollector{report: &HostReport{}, n: &cn},
		&fakeReporter{env: env, n: &rn},
		900*time.Second, quietLogger())
	obs := &recordingObserver{}
	loop.SetEnvelopeObserver(obs)

	got := loop.cycle(context.Background(), 900*time.Second)
	if len(obs.envs) != 1 || obs.envs[0].DesiredGeneration != 3 || !obs.envs[0].HasSignedOps {
		t.Fatalf("observer envelopes = %+v, want 1 with gen 3 + has_signed_ops", obs.envs)
	}
	if got != 120*time.Second {
		t.Errorf("poll interval = %v, want 120s adopted alongside the observer notify", got)
	}
}

// On a report failure the observer is NOT notified (there is no envelope to act on).
func TestLoop_ReportErrorSkipsObserver(t *testing.T) {
	var cn, rn int32
	loop := NewLoop(
		&fakeCollector{report: &HostReport{}, n: &cn},
		&fakeReporter{env: &ControlEnvelope{}, errSeq: []error{errors.New("hub 5xx")}, n: &rn},
		900*time.Second, quietLogger())
	obs := &recordingObserver{}
	loop.SetEnvelopeObserver(obs)
	loop.cycle(context.Background(), 900*time.Second)
	if len(obs.envs) != 0 {
		t.Errorf("observer notified on a report error: %+v", obs.envs)
	}
}

func TestClampInterval(t *testing.T) {
	cases := []struct {
		in      int
		wantSec int
		clamped bool
	}{
		{10, 60, true}, {59, 60, true}, {60, 60, false}, {120, 120, false},
		{3600, 3600, false}, {99999, 3600, true},
	}
	for _, c := range cases {
		d, clamped := clampInterval(c.in)
		if int(d.Seconds()) != c.wantSec || clamped != c.clamped {
			t.Errorf("clampInterval(%d) = %v,%v want %ds,%v", c.in, d, clamped, c.wantSec, c.clamped)
		}
	}
}

func TestLoop_CycleAdoptsAndClamps(t *testing.T) {
	current := 900 * time.Second
	mk := func(env *ControlEnvelope, collErr, repErr error) *Loop {
		var cn, rn int32
		return NewLoop(
			&fakeCollector{report: &HostReport{}, err: collErr, n: &cn},
			&fakeReporter{env: env, errSeq: []error{repErr}, n: &rn},
			current, quietLogger())
	}
	tests := []struct {
		name string
		env  *ControlEnvelope
		want time.Duration
	}{
		{"adopt in-range", &ControlEnvelope{PollIntervalSeconds: intPtr(120)}, 120 * time.Second},
		{"clamp low", &ControlEnvelope{PollIntervalSeconds: intPtr(10)}, 60 * time.Second},
		{"clamp high", &ControlEnvelope{PollIntervalSeconds: intPtr(99999)}, 3600 * time.Second},
		{"missing keeps current", &ControlEnvelope{PollIntervalSeconds: nil}, current},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := mk(tt.env, nil, nil).cycle(context.Background(), current)
			if got != tt.want {
				t.Errorf("cycle adopted %v, want %v", got, tt.want)
			}
		})
	}

	t.Run("collect error keeps current", func(t *testing.T) {
		got := mk(&ControlEnvelope{PollIntervalSeconds: intPtr(120)}, errors.New("x"), nil).cycle(context.Background(), current)
		if got != current {
			t.Errorf("got %v, want current %v", got, current)
		}
	})
	t.Run("report error keeps current", func(t *testing.T) {
		got := mk(nil, nil, errors.New("x")).cycle(context.Background(), current)
		if got != current {
			t.Errorf("got %v, want current %v", got, current)
		}
	})
}

func TestLoop_RunImmediateAndResilientAfterError(t *testing.T) {
	var cn, rn int32
	loop := NewLoop(
		&fakeCollector{report: &HostReport{}, n: &cn},
		// first report errors; subsequent ok; no interval override (keeps fast tick)
		&fakeReporter{env: &ControlEnvelope{}, errSeq: []error{errors.New("hub 5xx")}, n: &rn},
		10*time.Millisecond, quietLogger())

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- loop.Run(ctx) }()
	time.Sleep(90 * time.Millisecond)
	cancel()

	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("Run returned error: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("Run did not return after cancel")
	}
	// Immediate report + several ticks despite the first error → ≥3 collect calls.
	if got := atomic.LoadInt32(&cn); got < 3 {
		t.Errorf("collect calls = %d, want ≥3 (immediate + continuation after error)", got)
	}
}

func TestLoop_OutOfBandTriggerReportsImmediately(t *testing.T) {
	// The storage watchdog's trigger channel runs an extra report between ticks (the slow
	// cadence is 1h here, so any report within the window comes from the trigger).
	var cn, rn int32
	loop := NewLoop(
		&fakeCollector{report: &HostReport{}, n: &cn},
		&fakeReporter{env: &ControlEnvelope{}, n: &rn},
		time.Hour, quietLogger())
	trigger := make(chan struct{}, 1)
	loop.SetTrigger(trigger)

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- loop.Run(ctx) }()

	// Immediate report fires first (1 collect). Then fire the trigger → one more report.
	time.Sleep(20 * time.Millisecond)
	trigger <- struct{}{}
	time.Sleep(20 * time.Millisecond)
	cancel()
	<-done

	if got := atomic.LoadInt32(&cn); got < 2 {
		t.Errorf("collect calls = %d, want ≥2 (immediate + out-of-band trigger)", got)
	}
}

func TestLoop_RunAdoptsSlowerInterval(t *testing.T) {
	var cn, rn int32
	loop := NewLoop(
		&fakeCollector{report: &HostReport{}, n: &cn},
		// every report tells the agent to slow to 60s → after the immediate report,
		// the ticker resets to 60s and no further ticks fire within the test window.
		&fakeReporter{env: &ControlEnvelope{PollIntervalSeconds: intPtr(60)}, n: &rn},
		10*time.Millisecond, quietLogger())

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- loop.Run(ctx) }()
	time.Sleep(120 * time.Millisecond)
	cancel()
	<-done

	if got := atomic.LoadInt32(&cn); got != 1 {
		t.Errorf("collect calls = %d, want 1 (immediate report adopted 60s, ticker slowed)", got)
	}
}