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
}

// 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()`)

	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)
	lastUUID    map[string]string    // name -> last fs-UUID observed while ATTACHED (re-mount key)
}

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

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

// 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))
	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)})
		}
		// Re-mount candidate: a mount-backed target that is NOT mounted but whose backing
		// device is physically present (a disconnected→device-back state). Rate-limited per
		// target to the debounce window so a persistent mount failure can't storm HostOps.
		if w.remounter != nil && p.t.MountBacked && !p.present && p.devicePresent {
			if last, ok := w.lastRemount[p.t.Name]; !ok || now.Sub(last) >= w.debounce {
				w.lastRemount[p.t.Name] = now
				remounts = append(remounts, p.t)
			}
		}
		// Once a target is present again, clear its re-mount rate-limit so a future cycle
		// re-mounts promptly.
		if p.present {
			delete(w.lastRemount, 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 dispatching benign re-mount (device returned)",
			"target", t.Name, "where", t.MountPath)
		w.spawn(func() { w.remounter.Remount(ctx, t) })
	}
}

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
}