felhom-agent/internal/storage/watchdog.go

package storage

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

// Default watchdog timings (configurable via WatchdogOptions). The poll is FAST (seconds)
// so a USB drop is caught in seconds, not at the slow ~15-minute host-report cycle; the
// debounce keeps a flapping drive from storming the hub.
const (
	DefaultWatchdogInterval = 8 * time.Second
	DefaultWatchdogDebounce = 30 * time.Second
)

// KnownTarget is the watchdog's lightweight view of a target it watches. "Known" means a
// defined Proxmox storage (and/or a previously-observed-attached one); the watchdog only
// flags transitions for targets it has seen — it never reports a never-attached device.
type KnownTarget struct {
	Name          string
	Type          string
	DurableID     string
	UUID          string // fs-UUID (mount-backed targets) — the by-UUID re-mount key
	Network       bool   // nfs/cifs/pbs — liveness is a reachability dial, not a device check
	MountBacked   bool   // usb/local-dir — a drop = its mountpoint disappears
	BackingDevice string // resolved block device (local targets)
	MountPath     string // the mountpoint a mount-backed target must occupy
	ReachEndpoint string // host:port to dial for a network target's reachability
}

// KnownTargets enumerates the currently-known target set. Production wraps the Observer in
// CachingKnownTargets so the fast poll doesn't hammer the Proxmox API.
type KnownTargets interface {
	Known(ctx context.Context) ([]KnownTarget, error)
}

// TargetLiveness reports a known target's liveness. Production is HostLiveness (device/mount
// presence + a reachability dial, all non-privileged); tests inject a fake.
type TargetLiveness interface {
	// Present is the "in service" signal: mounted + reachable.
	Present(ctx context.Context, t KnownTarget) bool
	// DevicePresent is the "backing device is physically back" signal, independent of
	// whether it is mounted — the trigger for a benign re-mount of a returned drive.
	DevicePresent(ctx context.Context, t KnownTarget) bool
}

// Remounter performs the benign re-mount response when a known mount-backed target's device
// returns but its mountpoint is missing. The watchdog dispatches to it OFF its poll path
// (a goroutine), never synchronously under the lock. Production routes through the gate
// (benign) then HostOps.EnsureMount; wired in main.go so storage stays decoupled from
// reconcile. A nil Remounter disables the response (observe-only, Phase-A behaviour).
type Remounter interface {
	Remount(ctx context.Context, t KnownTarget)
}

// Transition is one observed state change for a known target (for logging/diagnostics).
type Transition struct {
	Name string
	From string // attached | disconnected
	To   string
}

// IntentReader gates self-heal by the drive's INTENT (slice 10 P3): the watchdog reconciles
// (re-mounts) ONLY a drive whose intent is `enrolled`. Satisfied by *IntentStore. When nil, the
// watchdog is ungated (legacy observe+remount-any behaviour) — production always wires it.
type IntentReader interface {
	Get(durableID string) DriveIntent
}

// Flapping guard (3C): a drive whose re-mount keeps not sticking gets exponential backoff, an alert
// after AlertThreshold consecutive failed cycles, and stops being retried after MaxRetries — instead
// of looping forever. Reset when the drive goes present (healed) or absent (gone, not flapping).
const (
	flappingAlertThreshold = 4 // consecutive failed re-mount cycles before raising the alert
	flappingMaxRetries     = 8 // after this many, stop retrying (alert stands until present/absent)
	flappingBackoffCap     = 5 // backoff window = debounce * 2^min(fails, cap)
)

// Watchdog is the third daemon goroutine (alongside the hub loop + reconcile engine). It
// fast-polls the known target set, detects attached↔disconnected transitions, and triggers
// an immediate, debounced out-of-band host-report so the hub learns of a drop in seconds.
//
// It NEVER mutates anything (Phase A is read-only) — the benign re-mount-by-UUID response
// to a return lands in Phase B. Here it only observes and signals.
type Watchdog struct {
	targets   KnownTargets
	liveness  TargetLiveness
	remounter Remounter // may be nil (observe-only)
	interval  time.Duration
	debounce  time.Duration
	trigger   func() // request an out-of-band report (debounced by the watchdog)
	logger    *slog.Logger
	now       func() time.Time
	spawn     func(func()) // spawn a background task (overridable in tests; default `go f()`)

	intent   IntentReader      // P3: gate re-mount by drive intent (nil = ungated legacy)
	onAbsent func(durableID string) // P3: called when a known target goes ABSENT (clears `ejected`)

	mu             sync.Mutex
	last           map[string]bool      // name -> last observed present (only for seen targets)
	lastFire       time.Time
	fired          bool                 // lastFire is valid
	pending        bool                 // a transition is awaiting the debounce window
	lastRemount    map[string]time.Time // name -> last re-mount dispatch (rate-limit / backoff base)
	lastUUID       map[string]string    // name -> last fs-UUID observed while ATTACHED (re-mount key)
	remountFails   map[string]int       // name -> consecutive failed re-mount cycles (flapping guard)
	remountPending map[string]bool      // name -> a re-mount was dispatched, awaiting present-confirm
	flapAlerted    map[string]bool      // name -> the "keeps dropping" alert has fired (once)
}

// WatchdogOptions configures a Watchdog. Targets, Liveness and Trigger are required; the
// rest default. Remounter is optional (nil = observe-only).
type WatchdogOptions struct {
	Targets   KnownTargets
	Liveness  TargetLiveness
	Remounter Remounter
	Trigger   func()
	Interval  time.Duration
	Debounce  time.Duration
	Logger    *slog.Logger
	// Intent gates self-heal to `enrolled` drives (slice 10 P3). OnAbsent is called (durable-id) when a
	// known target goes physically absent, so the intent store can clear an `ejected` flag (replug
	// rule). Both optional — nil Intent = ungated legacy remount; nil OnAbsent = no intent clearing.
	Intent   IntentReader
	OnAbsent func(durableID string)
}

// NewWatchdog builds a Watchdog. A nil Trigger is tolerated (the watchdog still tracks
// state, just signals nothing) so it degrades cleanly when no report sink is wired.
func NewWatchdog(opts WatchdogOptions) *Watchdog {
	interval := opts.Interval
	if interval <= 0 {
		interval = DefaultWatchdogInterval
	}
	debounce := opts.Debounce
	if debounce <= 0 {
		debounce = DefaultWatchdogDebounce
	}
	logger := opts.Logger
	if logger == nil {
		logger = slog.Default()
	}
	trigger := opts.Trigger
	if trigger == nil {
		trigger = func() {}
	}
	return &Watchdog{
		targets:        opts.Targets,
		liveness:       opts.Liveness,
		remounter:      opts.Remounter,
		intent:         opts.Intent,
		onAbsent:       opts.OnAbsent,
		interval:       interval,
		debounce:       debounce,
		trigger:        trigger,
		logger:         logger,
		now:            func() time.Time { return time.Now().UTC() },
		spawn:          func(f func()) { go f() },
		last:           map[string]bool{},
		lastRemount:    map[string]time.Time{},
		lastUUID:       map[string]string{},
		remountFails:   map[string]int{},
		remountPending: map[string]bool{},
		flapAlerted:    map[string]bool{},
	}
}

// reconcileAllowed reports whether self-heal may re-mount this target: only an `enrolled` drive is
// auto-mounted (3C: never auto-adopt a new/unknown drive; respect an intentional eject; never touch a
// decommissioned drive). An ungated watchdog (no intent reader) allows it (legacy).
func (w *Watchdog) reconcileAllowed(t KnownTarget) bool {
	if w.intent == nil {
		return true
	}
	return w.intent.Get(t.DurableID) == IntentEnrolled
}

// backoffWindow returns the re-mount backoff for a target with `fails` consecutive failed cycles:
// debounce * 2^min(fails, cap). fails==0 → the plain debounce.
func (w *Watchdog) backoffWindow(fails int) time.Duration {
	if fails > flappingBackoffCap {
		fails = flappingBackoffCap
	}
	win := w.debounce
	for i := 0; i < fails; i++ {
		win *= 2
	}
	return win
}

// Run fast-polls until ctx is cancelled. The first tick establishes the baseline (no
// trigger); subsequent ticks detect transitions. Returns nil on ctx cancellation.
func (w *Watchdog) Run(ctx context.Context) error {
	if w.targets == nil || w.liveness == nil {
		w.logger.Info("storage: watchdog idle (no target source / liveness probe configured)")
		<-ctx.Done()
		return nil
	}
	w.logger.Info("storage: watchdog starting", "interval", w.interval, "debounce", w.debounce)
	t := time.NewTicker(w.interval)
	defer t.Stop()
	w.tick(ctx) // immediate baseline
	for {
		select {
		case <-ctx.Done():
			w.logger.Info("storage: watchdog shutting down", "reason", ctx.Err())
			return nil
		case <-t.C:
			w.tick(ctx)
		}
	}
}

// tick performs one poll. Structure: probe liveness OUTSIDE the lock (the probes do IO —
// mount reads, dials), then take the lock only for the state diff + debounce decision, then
// perform side-effects (report trigger, re-mount dispatch) AFTER unlocking. The re-mount is
// handed to a background task — never run synchronously under the lock or on the poll path.
// Deterministic given w.now — tests drive it directly with a fake clock.
func (w *Watchdog) tick(ctx context.Context) {
	known, err := w.targets.Known(ctx)
	if err != nil {
		w.logger.Warn("storage: watchdog could not read known targets; skipping tick", "err", err)
		return
	}

	// Remember each target's fs-UUID while it is observable (attached), and backfill it
	// onto a target the current observe couldn't resolve (the unmounted case loses it). The
	// re-mount key is "sourced from the existing definition" (doc 03 §7): the agent learns
	// the UUID while attached, so a drop+return cycle can re-mount by-UUID even after the
	// known-set cache refreshed mid-drop. Single-goroutine (tick), guarded for -race.
	w.mu.Lock()
	for i := range known {
		if known[i].UUID != "" {
			w.lastUUID[known[i].Name] = known[i].UUID
		} else if u := w.lastUUID[known[i].Name]; u != "" {
			known[i].UUID = u
		}
	}
	w.mu.Unlock()

	// Probe outside the lock.
	type probe struct {
		t             KnownTarget
		present       bool
		devicePresent bool
	}
	probes := make([]probe, 0, len(known))
	for _, k := range known {
		p := probe{t: k, present: w.liveness.Present(ctx, k)}
		if k.MountBacked && !p.present {
			p.devicePresent = w.liveness.DevicePresent(ctx, k)
		}
		probes = append(probes, p)
	}

	now := w.now()

	w.mu.Lock()
	var transitions []Transition
	var remounts []KnownTarget
	current := make(map[string]bool, len(probes))
	var absentNow []KnownTarget  // P3: targets that just went physically ABSENT (clear `ejected`)
	var flapAlerts []KnownTarget // P3: "drive keeps dropping" alerts to surface
	for _, p := range probes {
		current[p.t.Name] = p.present
		if prev, seen := w.last[p.t.Name]; seen && prev != p.present {
			transitions = append(transitions, Transition{Name: p.t.Name, From: stateStr(prev), To: stateStr(p.present)})
			// Physically ABSENT (device gone, not merely unmounted): a mount-backed target that went
			// !present AND whose device is no longer present. The intent store clears `ejected` here so
			// a replug auto-mounts (the replug rule); flapping state resets (it's gone, not flapping).
			if prev && !p.present && p.t.MountBacked && !p.devicePresent {
				absentNow = append(absentNow, p.t)
				w.resetFlap(p.t.Name)
			}
		}
		// Re-mount candidate (self-heal): an ENROLLED, mount-backed target that is NOT mounted but
		// whose backing device IS physically present (out-of-band unmount → device-still-there). Gated
		// by intent (never auto-adopt a new/ejected/decommissioned drive) + an exponential-backoff
		// flapping guard (3C): a re-mount that doesn't stick (still !present next cycle) counts as a
		// failure → grow the backoff; alert after AlertThreshold; stop after MaxRetries.
		if w.remounter != nil && p.t.MountBacked && !p.present && p.devicePresent && w.reconcileAllowed(p.t) {
			fails := w.remountFails[p.t.Name]
			// Only act once a full backoff window has elapsed since the last dispatch (or the first
			// time). Gating the WHOLE evaluation on the window means a re-mount that's merely slow
			// (the dispatch is async, ticks are fast) isn't miscounted as a failure — a failure is
			// "still not present a full window after we dispatched".
			if last, ok := w.lastRemount[p.t.Name]; !ok || now.Sub(last) >= w.backoffWindow(fails) {
				if w.remountPending[p.t.Name] { // prior dispatch didn't take within its window → failed
					fails++
					w.remountFails[p.t.Name] = fails
					w.remountPending[p.t.Name] = false
				}
				if fails >= flappingAlertThreshold && !w.flapAlerted[p.t.Name] {
					w.flapAlerted[p.t.Name] = true
					flapAlerts = append(flapAlerts, p.t)
				}
				if fails < flappingMaxRetries {
					w.lastRemount[p.t.Name] = now
					w.remountPending[p.t.Name] = true
					remounts = append(remounts, p.t)
				}
			}
		}
		// Healed (present again): reset re-mount rate-limit + flapping state so a future drop re-mounts
		// promptly and the alert can re-fire.
		if p.present {
			w.resetFlap(p.t.Name)
		}
	}
	w.last = current // targets no longer known drop out

	doFire := false
	if len(transitions) > 0 {
		if !w.fired || now.Sub(w.lastFire) >= w.debounce {
			doFire = true
			w.lastFire, w.fired, w.pending = now, true, false
		} else {
			w.pending = true
		}
	} else if w.pending && now.Sub(w.lastFire) >= w.debounce {
		doFire = true
		w.lastFire, w.pending = now, false
	}
	w.mu.Unlock()

	// Side-effects, off the lock.
	for _, tr := range transitions {
		w.logger.Warn("storage: watchdog detected target state change",
			"target", tr.Name, "from", tr.From, "to", tr.To)
	}
	if doFire {
		w.logger.Info("storage: watchdog triggering out-of-band host-report", "transitions", len(transitions))
		w.trigger()
	}
	for _, t := range remounts {
		t := t
		w.logger.Info("storage: watchdog self-heal — re-mounting enrolled drive (device returned, no eject intent)",
			"target", t.Name, "where", t.MountPath, "durable_id", t.DurableID)
		w.spawn(func() { w.remounter.Remount(ctx, t) })
	}
	// P3: a drive that went physically absent — clear an `ejected` intent so a replug auto-mounts.
	for _, t := range absentNow {
		if w.onAbsent != nil && t.DurableID != "" {
			w.logger.Info("storage: watchdog — known drive went absent; clearing any eject intent (replug will auto-mount)",
				"target", t.Name, "durable_id", t.DurableID)
			w.onAbsent(t.DurableID)
		}
	}
	// P3 flapping guard: a drive whose re-mount keeps not sticking — alert instead of looping silently.
	for _, t := range flapAlerts {
		w.logger.Warn("storage: watchdog ALERT — drive keeps dropping (re-mount not sticking); backing off",
			"target", t.Name, "where", t.MountPath, "durable_id", t.DurableID, "failed_cycles", flappingAlertThreshold)
		w.trigger() // surface out-of-band so the hub/operator sees the unhealthy drive
	}
}

// resetFlap clears all per-target re-mount/backoff/alert state (target healed or went absent). The
// caller holds w.mu.
func (w *Watchdog) resetFlap(name string) {
	delete(w.lastRemount, name)
	delete(w.remountFails, name)
	delete(w.remountPending, name)
	delete(w.flapAlerted, name)
}

func stateStr(present bool) string {
	if present {
		return "attached"
	}
	return "disconnected"
}

// --- production liveness + a caching known-target source ---

// HostLiveness is the production TargetLiveness: device + mount presence for local
// targets (the fast USB-drop signal) and a short reachability dial for network targets.
// All non-privileged.
type HostLiveness struct {
	host        HostReader
	dialTimeout time.Duration
	dial        func(network, address string, timeout time.Duration) (net.Conn, error)
}

// NewHostLiveness builds a HostLiveness over a HostReader. dialTimeout defaults to 3s.
func NewHostLiveness(host HostReader, dialTimeout time.Duration) *HostLiveness {
	if host == nil {
		host = NewProcHostReader()
	}
	if dialTimeout <= 0 {
		dialTimeout = 3 * time.Second
	}
	return &HostLiveness{host: host, dialTimeout: dialTimeout, dial: net.DialTimeout}
}

// Present probes one target without touching Proxmox.
func (h *HostLiveness) Present(ctx context.Context, t KnownTarget) bool {
	if t.Network {
		if t.ReachEndpoint == "" {
			return true // can't probe → don't false-alarm; the 15-min cycle uses the active flag
		}
		conn, err := h.dial("tcp", t.ReachEndpoint, h.dialTimeout)
		if err != nil {
			return false
		}
		_ = conn.Close()
		return true
	}
	if t.MountBacked {
		// A mount-backed target (USB / extra disk) is present iff its mountpoint is an
		// active mount AND the backing device node exists.
		if !h.mounted(t.MountPath) {
			return false
		}
		return t.BackingDevice == "" || h.host.DeviceExists(t.BackingDevice)
	}
	// Non-removable builtin targets (local/lvmthin): treated as present here — they don't
	// "drop" without the whole host going down, which the heartbeat covers.
	return true
}

// DevicePresent reports whether the backing device is physically present (regardless of
// mount state) — the re-mount trigger. Checks /dev/disk/by-uuid/<UUID> first (the by-UUID
// link appears when the drive is plugged), then any known backing-device node.
func (h *HostLiveness) DevicePresent(ctx context.Context, t KnownTarget) bool {
	if !t.MountBacked {
		return false
	}
	if t.UUID != "" {
		if dev, err := ByUUIDDevicePath(t.UUID); err == nil && h.host.DeviceExists(dev) {
			return true
		}
	}
	return t.BackingDevice != "" && h.host.DeviceExists(t.BackingDevice)
}

func (h *HostLiveness) mounted(path string) bool {
	if path == "" {
		return false
	}
	mounts, err := h.host.Mounts()
	if err != nil {
		return false
	}
	_, _, ok := exactMountDevice(mounts, path)
	return ok
}

// CachingKnownTargets wraps a slow KnownTargets source (the Observer, which hits Proxmox)
// with a TTL so the fast watchdog poll re-derives the known SET only every ttl, while
// still probing liveness every tick. A read error returns the last good set (so a
// transient Proxmox blip doesn't blank the watchdog's world).
type CachingKnownTargets struct {
	src    KnownTargets
	ttl    time.Duration
	now    func() time.Time
	mu     sync.Mutex
	cached []KnownTarget
	at     time.Time
	loaded bool
}

// NewCachingKnownTargets wraps src, refreshing at most every ttl (default 60s).
func NewCachingKnownTargets(src KnownTargets, ttl time.Duration) *CachingKnownTargets {
	if ttl <= 0 {
		ttl = 60 * time.Second
	}
	return &CachingKnownTargets{src: src, ttl: ttl, now: func() time.Time { return time.Now().UTC() }}
}

// Known returns the cached set, refreshing it when the TTL has elapsed.
func (c *CachingKnownTargets) Known(ctx context.Context) ([]KnownTarget, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	now := c.now()
	if c.loaded && now.Sub(c.at) < c.ttl {
		return c.cached, nil
	}
	fresh, err := c.src.Known(ctx)
	if err != nil {
		if c.loaded {
			return c.cached, nil // serve stale rather than blank on a transient error
		}
		return nil, err
	}
	c.cached, c.at, c.loaded = fresh, now, true
	return c.cached, nil
}