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
	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 whether one known target is presently up. Production is
// HostLiveness (device/mount presence + a reachability dial, all non-privileged); tests
// inject a fake.
type TargetLiveness interface {
	Present(ctx context.Context, t KnownTarget) bool
}

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

	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
}

// WatchdogOptions configures a Watchdog. Targets, Liveness and Trigger are required; the
// rest default.
type WatchdogOptions struct {
	Targets  KnownTargets
	Liveness TargetLiveness
	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,
		interval: interval,
		debounce: debounce,
		trigger:  trigger,
		logger:   logger,
		now:      func() time.Time { return time.Now().UTC() },
		last:     map[string]bool{},
	}
}

// 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: read the known set, probe each target's liveness, diff against
// the last-seen state, and fire a debounced trigger on any transition for a SEEN target.
// It is 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
	}

	w.mu.Lock()
	defer w.mu.Unlock()

	var transitions []Transition
	current := make(map[string]bool, len(known))
	for _, k := range known {
		present := w.liveness.Present(ctx, k)
		current[k.Name] = present
		prev, seen := w.last[k.Name]
		if !seen {
			continue // first observation → baseline only (never flag a never-attached drop)
		}
		if prev != present {
			transitions = append(transitions, Transition{Name: k.Name, From: stateStr(prev), To: stateStr(present)})
		}
	}
	// Replace the baseline with the current snapshot (targets no longer known drop out).
	w.last = current

	now := w.now()
	if len(transitions) > 0 {
		for _, tr := range transitions {
			w.logger.Warn("storage: watchdog detected target state change",
				"target", tr.Name, "from", tr.From, "to", tr.To)
		}
		if !w.fired || now.Sub(w.lastFire) >= w.debounce {
			w.fire(now, len(transitions))
		} else {
			w.pending = true
			w.logger.Debug("storage: watchdog debouncing transition", "pending_until", w.lastFire.Add(w.debounce))
		}
		return
	}
	// No new transition, but a debounced one is pending and the window has elapsed → fire.
	if w.pending && now.Sub(w.lastFire) >= w.debounce {
		w.fire(now, 0)
	}
}

// fire requests the out-of-band report and resets the debounce window. Called under w.mu.
func (w *Watchdog) fire(now time.Time, n int) {
	w.lastFire = now
	w.fired = true
	w.pending = false
	w.logger.Info("storage: watchdog triggering out-of-band host-report", "transitions", n)
	w.trigger()
}

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
}

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
}