felhom-agent/internal/storage/watchdog_test.go

package storage

import (
	"context"
	"errors"
	"net"
	"path/filepath"
	"sync"
	"testing"
	"time"
)

// staticKnown is a settable KnownTargets fake.
type staticKnown struct {
	mu      sync.Mutex
	targets []KnownTarget
	err     error
	calls   int
}

func (s *staticKnown) Known(context.Context) ([]KnownTarget, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.calls++
	return s.targets, s.err
}

// mapLiveness is a settable per-target presence + device-presence fake.
type mapLiveness struct {
	mu      sync.Mutex
	present map[string]bool
	device  map[string]bool // backing-device presence (re-mount trigger)
}

func (m *mapLiveness) set(name string, p bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.present[name] = p
}
func (m *mapLiveness) setDevice(name string, p bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.device == nil {
		m.device = map[string]bool{}
	}
	m.device[name] = p
}
func (m *mapLiveness) Present(_ context.Context, t KnownTarget) bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.present[t.Name]
}
func (m *mapLiveness) DevicePresent(_ context.Context, t KnownTarget) bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.device[t.Name]
}

// newTestWatchdog builds a watchdog with a manual clock and a trigger counter.
func newTestWatchdog(known KnownTargets, live TargetLiveness, debounce time.Duration) (*Watchdog, *int, *time.Time) {
	var fires int
	clock := time.Unix(1_700_000_000, 0).UTC()
	w := NewWatchdog(WatchdogOptions{
		Targets:  known,
		Liveness: live,
		Trigger:  func() { fires++ },
		Interval: time.Second,
		Debounce: debounce,
		Logger:   quietLogger(),
	})
	w.now = func() time.Time { return clock }
	return w, &fires, &clock
}

func TestWatchdog_BaselineThenDropTriggers(t *testing.T) {
	known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true}}}
	live := &mapLiveness{present: map[string]bool{"usb": true}}
	w, fires, _ := newTestWatchdog(known, live, 30*time.Second)
	ctx := context.Background()

	w.tick(ctx) // baseline: present, no trigger
	if *fires != 0 {
		t.Fatalf("baseline tick must not trigger, fires=%d", *fires)
	}
	live.set("usb", false) // drop
	w.tick(ctx)
	if *fires != 1 {
		t.Fatalf("a known target drop must trigger an out-of-band report, fires=%d", *fires)
	}
}

func TestWatchdog_NeverAttachedNotFlagged(t *testing.T) {
	// A defined-but-absent target (never seen present) must not be flagged on its absence.
	known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true}}}
	live := &mapLiveness{present: map[string]bool{"usb": false}}
	w, fires, _ := newTestWatchdog(known, live, 30*time.Second)
	ctx := context.Background()

	w.tick(ctx) // baseline absent
	w.tick(ctx) // still absent
	if *fires != 0 {
		t.Fatalf("a never-attached target must not trigger, fires=%d", *fires)
	}
	// Now it appears (reconnect) → that IS a transition worth reporting.
	live.set("usb", true)
	w.tick(ctx)
	if *fires != 1 {
		t.Fatalf("attach transition should trigger, fires=%d", *fires)
	}
}

func TestWatchdog_DebounceCoalescesFlaps(t *testing.T) {
	known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true}}}
	live := &mapLiveness{present: map[string]bool{"usb": true}}
	w, fires, clock := newTestWatchdog(known, live, 30*time.Second)
	ctx := context.Background()

	w.tick(ctx) // baseline present

	// First drop fires immediately (leading edge).
	live.set("usb", false)
	w.tick(ctx)
	if *fires != 1 {
		t.Fatalf("first drop should fire, fires=%d", *fires)
	}

	// Flap within the debounce window: re-attach then drop again — suppressed (pending).
	*clock = clock.Add(5 * time.Second)
	live.set("usb", true)
	w.tick(ctx)
	*clock = clock.Add(5 * time.Second)
	live.set("usb", false)
	w.tick(ctx)
	if *fires != 1 {
		t.Fatalf("flaps within the debounce window must be coalesced, fires=%d", *fires)
	}

	// After the window elapses, the pending change fires (trailing edge), even with no new
	// transition this tick.
	*clock = clock.Add(30 * time.Second)
	w.tick(ctx)
	if *fires != 2 {
		t.Fatalf("a pending change must fire once the window elapses, fires=%d", *fires)
	}
}

// fakeRemounter records re-mount dispatches.
type fakeRemounter struct {
	mu    sync.Mutex
	calls []string
	uuids []string
}

func (r *fakeRemounter) Remount(_ context.Context, t KnownTarget) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.calls = append(r.calls, t.Name)
	r.uuids = append(r.uuids, t.UUID)
}
func (r *fakeRemounter) count() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.calls)
}

func TestWatchdog_ReMountOnDeviceReturn(t *testing.T) {
	known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true, UUID: "1234-ABCD", MountPath: "/mnt/usb"}}}
	live := &mapLiveness{present: map[string]bool{"usb": true}, device: map[string]bool{"usb": true}}
	rem := &fakeRemounter{}
	w, _, clock := newTestWatchdog(known, live, 30*time.Second)
	w.remounter = rem
	w.spawn = func(f func()) { f() } // run the dispatch synchronously for deterministic assertion
	ctx := context.Background()

	w.tick(ctx) // baseline: present
	if rem.count() != 0 {
		t.Fatalf("no re-mount at baseline, got %d", rem.count())
	}

	// Drop: device gone, unmounted. No re-mount (nothing to mount).
	live.set("usb", false)
	live.setDevice("usb", false)
	w.tick(ctx)
	if rem.count() != 0 {
		t.Fatalf("no re-mount while device absent, got %d", rem.count())
	}

	// Device returns but still unmounted → re-mount dispatched.
	live.setDevice("usb", true)
	w.tick(ctx)
	if rem.count() != 1 {
		t.Fatalf("re-mount expected when device returns unmounted, got %d", rem.count())
	}

	// Still device-present-unmounted within the debounce window → rate-limited (no storm).
	*clock = clock.Add(5 * time.Second)
	w.tick(ctx)
	if rem.count() != 1 {
		t.Fatalf("re-mount must be rate-limited within debounce, got %d", rem.count())
	}

	// Successful mount (present=true) clears the rate-limit; a later cycle re-mounts again.
	live.set("usb", true)
	*clock = clock.Add(5 * time.Second)
	w.tick(ctx)            // present → clears lastRemount
	live.set("usb", false) // drop again, device still present
	*clock = clock.Add(5 * time.Second)
	w.tick(ctx)
	if rem.count() != 2 {
		t.Fatalf("a fresh device cycle should re-mount again, got %d", rem.count())
	}
}

func TestWatchdog_ReMountUsesRememberedUUID(t *testing.T) {
	// The re-mount key must survive the known-set cache losing the UUID mid-drop (an
	// unmounted dir-storage can't resolve its own UUID). The watchdog remembers the UUID
	// observed while attached and backfills it onto the re-mount target.
	known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true, UUID: "277a2179", MountPath: "/mnt/usb"}}}
	live := &mapLiveness{present: map[string]bool{"usb": true}, device: map[string]bool{"usb": true}}
	rem := &fakeRemounter{}
	w, _, _ := newTestWatchdog(known, live, 30*time.Second)
	w.remounter = rem
	w.spawn = func(f func()) { f() }
	ctx := context.Background()

	w.tick(ctx) // baseline attached → remembers UUID 277a2179

	// Cache refreshes during the drop and loses the UUID (the unmounted observe can't
	// resolve it); device still present, unmounted.
	known.targets = []KnownTarget{{Name: "usb", MountBacked: true, UUID: "", MountPath: "/mnt/usb"}}
	live.set("usb", false)
	w.tick(ctx)

	if rem.count() != 1 {
		t.Fatalf("expected one re-mount, got %d", rem.count())
	}
	if rem.uuids[0] != "277a2179" {
		t.Fatalf("re-mount must use the remembered UUID, got %q", rem.uuids[0])
	}
}

func TestWatchdog_ReadErrorSkipsTick(t *testing.T) {
	known := &staticKnown{err: errors.New("proxmox blip")}
	live := &mapLiveness{present: map[string]bool{}}
	w, fires, _ := newTestWatchdog(known, live, time.Second)
	w.tick(context.Background())
	if *fires != 0 {
		t.Fatalf("a known-target read error must not trigger, fires=%d", *fires)
	}
}

func TestWatchdog_RunBaselinesAndStops(t *testing.T) {
	// Smoke test of the goroutine wiring under -race: Run establishes a baseline and exits
	// cleanly on ctx cancellation.
	known := &staticKnown{targets: []KnownTarget{{Name: "usb", MountBacked: true}}}
	live := &mapLiveness{present: map[string]bool{"usb": true}}
	w, _, _ := newTestWatchdog(known, live, time.Second)
	w.now = func() time.Time { return time.Now().UTC() }
	w.interval = 5 * time.Millisecond

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- w.Run(ctx) }()
	time.Sleep(30 * time.Millisecond)
	cancel()
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("Run returned %v, want nil on cancel", err)
		}
	case <-time.After(time.Second):
		t.Fatal("watchdog did not stop on cancel")
	}
}

func TestCachingKnownTargets_RefreshesOnTTL(t *testing.T) {
	src := &staticKnown{targets: []KnownTarget{{Name: "a"}}}
	clock := time.Unix(1_700_000_000, 0).UTC()
	c := NewCachingKnownTargets(src, 60*time.Second)
	c.now = func() time.Time { return clock }
	ctx := context.Background()

	if _, err := c.Known(ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := c.Known(ctx); err != nil { // within TTL → cached
		t.Fatal(err)
	}
	if src.calls != 1 {
		t.Fatalf("within TTL the source must be hit once, calls=%d", src.calls)
	}
	clock = clock.Add(61 * time.Second) // past TTL
	if _, err := c.Known(ctx); err != nil {
		t.Fatal(err)
	}
	if src.calls != 2 {
		t.Fatalf("past TTL the source must refresh, calls=%d", src.calls)
	}
}

func TestCachingKnownTargets_ServesStaleOnError(t *testing.T) {
	src := &staticKnown{targets: []KnownTarget{{Name: "a"}}}
	clock := time.Unix(1_700_000_000, 0).UTC()
	c := NewCachingKnownTargets(src, 1*time.Second)
	c.now = func() time.Time { return clock }
	ctx := context.Background()

	if _, err := c.Known(ctx); err != nil { // prime the cache
		t.Fatal(err)
	}
	clock = clock.Add(2 * time.Second)
	src.mu.Lock()
	src.err = errors.New("blip")
	src.mu.Unlock()
	got, err := c.Known(ctx)
	if err != nil {
		t.Fatalf("a transient error must serve stale, got err=%v", err)
	}
	if len(got) != 1 || got[0].Name != "a" {
		t.Fatalf("stale set not served: %+v", got)
	}
}

func TestHostLiveness_MountBackedPresence(t *testing.T) {
	host := &fakeHostReader{
		mounts: []Mount{{Device: "/dev/sdb1", MountPoint: "/mnt/usb", FSType: "ext4"}},
		exists: map[string]bool{"/dev/sdb1": true},
	}
	hl := NewHostLiveness(host, time.Second)
	tgt := KnownTarget{Name: "usb", MountBacked: true, MountPath: "/mnt/usb", BackingDevice: "/dev/sdb1"}
	if !hl.Present(context.Background(), tgt) {
		t.Error("mounted device should be present")
	}

	// Unmount it: no exact mount entry → absent.
	host.mounts = []Mount{{Device: "/dev/mapper/root", MountPoint: "/", FSType: "ext4"}}
	if hl.Present(context.Background(), tgt) {
		t.Error("unmounted device should be absent")
	}
}

func TestHostLiveness_NetworkDial(t *testing.T) {
	hl := NewHostLiveness(&fakeHostReader{}, time.Second)
	var dialed string
	hl.dial = func(network, addr string, _ time.Duration) (net.Conn, error) {
		dialed = addr
		return nil, errors.New("refused")
	}
	tgt := KnownTarget{Name: "nfs", Network: true, ReachEndpoint: "10.0.0.5:2049"}
	if hl.Present(context.Background(), tgt) {
		t.Error("a refused dial should report not-present")
	}
	if dialed != "10.0.0.5:2049" {
		t.Errorf("dialed %q, want 10.0.0.5:2049", dialed)
	}

	// No endpoint to probe → don't false-alarm (the slow cycle uses the active flag).
	if !hl.Present(context.Background(), KnownTarget{Name: "x", Network: true}) {
		t.Error("network target without endpoint must not be flagged down")
	}
}

// ---- P3 self-heal reconcile (intent-gated) -----------------------------------------------

// newReconcileWD builds a watchdog with one mount-backed, durable-id'd target ("usb"), a fake
// remounter, an intent store, and the OnAbsent hook wired to it. Synchronous dispatch + manual clock.
func newReconcileWD(intent *IntentStore) (*Watchdog, *mapLiveness, *fakeRemounter, *time.Time) {
	known := &staticKnown{targets: []KnownTarget{{
		Name: "usb", DurableID: "uuid:U", MountBacked: true,
		MountPath: "/mnt/usb", BackingDevice: "/dev/sdb1", UUID: "U",
	}}}
	live := &mapLiveness{present: map[string]bool{}, device: map[string]bool{}}
	rem := &fakeRemounter{}
	clock := time.Unix(1_700_000_000, 0).UTC()
	var reader IntentReader
	if intent != nil {
		reader = intent
	}
	w := NewWatchdog(WatchdogOptions{
		Targets: known, Liveness: live, Remounter: rem, Intent: reader,
		OnAbsent: func(id string) {
			if intent != nil {
				_ = intent.OnAbsent(id)
			}
		},
		Interval: time.Second, Debounce: 30 * time.Second, Logger: quietLogger(),
	})
	w.now = func() time.Time { return clock }
	w.spawn = func(f func()) { f() }
	return w, live, rem, &clock
}

// colleague's out-of-band unmount (intent=enrolled, device present, not mounted) → reconciled.
func TestWatchdog_Reconcile_EnrolledColleagueUnmount(t *testing.T) {
	intent, _ := OpenIntentStore(filepath.Join(t.TempDir(), "i.json"))
	_ = intent.SetEnrolled("uuid:U")
	w, live, rem, _ := newReconcileWD(intent)
	ctx := context.Background()
	live.set("usb", true)
	w.tick(ctx) // baseline present
	live.set("usb", false)
	live.setDevice("usb", true) // unmounted, device still there (out-of-band unmount)
	w.tick(ctx)
	if rem.count() != 1 {
		t.Fatalf("enrolled drive out-of-band-unmounted should be reconciled: remounts=%d", rem.count())
	}
}

// ejected / new / decommissioned → NEVER reconciled (only enrolled is).
func TestWatchdog_Reconcile_RespectsIntent(t *testing.T) {
	for _, tc := range []struct {
		name  string
		setup func(*IntentStore)
	}{
		{"ejected", func(s *IntentStore) { _ = s.SetEjected("uuid:U") }},
		{"new", func(s *IntentStore) {}}, // no record
		{"decommissioned", func(s *IntentStore) { _ = s.SetDecommissioned("uuid:U") }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			intent, _ := OpenIntentStore(filepath.Join(t.TempDir(), "i.json"))
			tc.setup(intent)
			w, live, rem, _ := newReconcileWD(intent)
			ctx := context.Background()
			live.set("usb", true)
			w.tick(ctx)
			live.set("usb", false)
			live.setDevice("usb", true)
			w.tick(ctx)
			if rem.count() != 0 {
				t.Fatalf("%s drive must NOT be reconciled: remounts=%d", tc.name, rem.count())
			}
		})
	}
}

// ejected → physically absent → (OnAbsent clears to enrolled) → replug → reconciled (replug rule).
func TestWatchdog_Reconcile_EjectedAbsentReplugAutoMounts(t *testing.T) {
	intent, _ := OpenIntentStore(filepath.Join(t.TempDir(), "i.json"))
	_ = intent.SetEjected("uuid:U")
	w, live, rem, _ := newReconcileWD(intent)
	ctx := context.Background()
	live.set("usb", true)
	w.tick(ctx) // baseline present
	// physically pull: not present AND device gone
	live.set("usb", false)
	live.setDevice("usb", false)
	w.tick(ctx) // present→absent, device gone → OnAbsent clears ejected→enrolled
	if intent.Get("uuid:U") != IntentEnrolled {
		t.Fatalf("ejected→absent should clear to enrolled, got %q", intent.Get("uuid:U"))
	}
	// replug: device back, not yet mounted
	live.setDevice("usb", true)
	w.tick(ctx)
	if rem.count() != 1 {
		t.Fatalf("replugged drive (now enrolled) should auto-mount: remounts=%d", rem.count())
	}
}

// flapping: a re-mount that never sticks backs off and STOPS after MaxRetries (no infinite loop).
func TestWatchdog_Reconcile_FlappingBacksOffAndCaps(t *testing.T) {
	intent, _ := OpenIntentStore(filepath.Join(t.TempDir(), "i.json"))
	_ = intent.SetEnrolled("uuid:U")
	w, live, rem, clock := newReconcileWD(intent)
	ctx := context.Background()
	live.set("usb", true)
	w.tick(ctx) // baseline present
	// Drive is stuck: device present but the re-mount never makes it present.
	live.set("usb", false)
	live.setDevice("usb", true)
	for i := 0; i < 20; i++ {
		w.tick(ctx)
		*clock = clock.Add(2 * time.Hour) // always exceed the (growing) backoff window
	}
	if rem.count() != flappingMaxRetries {
		t.Fatalf("flapping re-mount should cap at %d, got %d (infinite loop?)", flappingMaxRetries, rem.count())
	}
	// further ticks add no more re-mounts (stays capped)
	w.tick(ctx)
	if rem.count() != flappingMaxRetries {
		t.Fatalf("capped flapping must not resume: got %d", rem.count())
	}
}