felhom-controller/controller/internal/channelhealth/checker.go

// Package channelhealth periodically proves the controller→agent local-API channel and surfaces a
// classified operator alert + dashboard entry on a state change. It closes the gap the R1
// pin-mismatch incident exposed: the channel was only probed once at startup (probeLocalAPI) and only
// logged, so a mid-life agent re-key went unnoticed until a user hit the dead disk UI.
//
// Design is spike-driven (SPIKE-controller-agent-channel-health-2026-06-29, GO):
//   - Probe via the PRODUCTION memoized agent client (reused, not a fresh one): it self-heals after an
//     agent restart, reflects exactly what the UI sees, and avoids the per-call transport leak the
//     singleton fixed. The construction error (bad fingerprint → agentClient() fails, latching via
//     sync.Once) is surfaced distinctly.
//   - Classify the failure by the error substring (the spike Q1 map).
//   - Debounce transient reasons (connection-refused / timeout): a clean agent restart shows a ~1s
//     socket gap (one failed probe) that must NOT page — require N>=2 consecutive before alerting.
//     Pin-mismatch / 401 / construction / DNS alert on first observation (they don't self-clear).
//   - First scheduler observation SEEDS state without notifying (mirrors host_staleness/host_capability).
package channelhealth

import (
	"context"
	"log"
	"strings"
	"sync"
	"time"
)

// debounceThreshold is the number of consecutive down-probes a TRANSIENT reason needs before it is
// treated as a real outage (suppresses the ~1s agent-restart blip from the spike's Q2).
const debounceThreshold = 2

// probeTimeout mirrors probeLocalAPI's 15s budget for the GET /storage call.
const probeTimeout = 15 * time.Second

// Reason is the classified channel-down cause.
type Reason string

const (
	ReasonPinMismatch   Reason = "pin_mismatch"
	ReasonUnauthorized  Reason = "unauthorized"
	ReasonUnreachable   Reason = "unreachable"
	ReasonTimeout       Reason = "timeout"
	ReasonMisconfigured Reason = "misconfigured"      // endpoint unresolvable (DNS)
	ReasonConstruction  Reason = "construction_error" // agentClient() build error — LATCHES until restart
	ReasonUnknown       Reason = "unknown"
)

// Probe runs one channel check via the production memoized client. It returns whether the failure was
// a CONSTRUCTION error (agentClient() couldn't even build — a latching config fault) and the error
// (nil = channel up). Production wires this to Server.ProbeAgentChannel; tests inject a fake.
type Probe func(ctx context.Context) (constructionErr bool, err error)

// Sink receives the checker's outputs. SetDashboard reflects the CURRENT state every check (idempotent
// — a born-down channel shows immediately, with no notification). NotifyDown/NotifyRecovered fire ONLY
// on a transition (the operator relay, with the hub-side cooldown). Implemented in main.go over the
// notify.Notifier + web.AlertManager (kept out of this package to avoid an import cycle).
type Sink interface {
	SetDashboard(down bool, reason Reason, hungarianMsg string)
	NotifyDown(reason Reason, eventType, severity, englishMsg string)
	NotifyRecovered()
}

type classification struct {
	reason    Reason
	eventType string
	severity  string // "error" (critical) | "warning" (transient)
	english   string // operator alert message (relayed to the hub)
	hungarian string // short dashboard line
	debounce  bool   // transient → require N>=2 consecutive
}

// classify maps a probe result to a classification. constructionErr is authoritative for the latching
// config fault; otherwise the error substring decides (spike Q1 map). Match substrings, not exact
// strings (the wrapped Get "..." prefix varies).
func classify(constructionErr bool, err error) classification {
	if constructionErr {
		return classification{ReasonConstruction, "agent_channel_construction_error", "error",
			"Controller→agent channel down: agent client misconfigured — local_api.fingerprint is not a valid SHA-256, the controller cannot build the client (config fix + controller restart needed).",
			"A tárolókezelő ügynök kapcsolata hibásan beállítva.", false}
	}
	s := err.Error()
	switch {
	case strings.Contains(s, "TLS pin mismatch"):
		return classification{ReasonPinMismatch, "agent_channel_pin_mismatch", "error",
			"Controller→agent channel down: TLS pin mismatch — the agent's leaf cert no longer matches the controller's bootstrap fingerprint (re-pin / re-bootstrap).",
			"A tárolókezelő ügynök tanúsítványa megváltozott.", false}
	case strings.Contains(s, "HTTP 401"):
		return classification{ReasonUnauthorized, "agent_channel_unauthorized", "error",
			"Controller→agent channel down: agent rejected the controller token (HTTP 401) — token stale/rotated (re-bootstrap).",
			"A tárolókezelő ügynök elutasította a hozzáférést.", false}
	case strings.Contains(s, "no such host") || strings.Contains(s, "lookup "):
		return classification{ReasonMisconfigured, "agent_channel_misconfigured", "error",
			"Controller→agent channel down: agent endpoint unresolvable — local_api.endpoint misconfigured.",
			"A tárolókezelő ügynök címe nem feloldható.", false}
	case strings.Contains(s, "connection refused"):
		return classification{ReasonUnreachable, "agent_channel_unreachable", "warning",
			"Controller→agent channel down: agent unreachable (connection refused) — felhom-agent down or :8443 closed.",
			"A tárolókezelő ügynök nem elérhető.", true}
	case strings.Contains(s, "context deadline exceeded") || strings.Contains(s, "i/o timeout") || strings.Contains(s, "Client.Timeout"):
		return classification{ReasonTimeout, "agent_channel_timeout", "warning",
			"Controller→agent channel down: timeout — host unreachable / packets dropped.",
			"A tárolókezelő ügynök nem válaszol.", true}
	default:
		return classification{ReasonUnknown, "agent_channel_unknown", "warning",
			"Controller→agent channel down: " + s,
			"A tárolókezelő ügynök nem elérhető.", true}
	}
}

// Checker holds the in-memory channel state. No persistence — the state is re-derived each run
// (mirrors the AlertManager's state-based model). Safe for the single scheduler caller; the mutex
// guards against an overlapping run.
type Checker struct {
	probe  Probe
	sink   Sink
	logger *log.Logger

	mu              sync.Mutex
	state           string // "" (unseeded) | "up" | "down:<reason>"
	consecutiveDown int
	alerted         bool // have we emitted a down alert for the CURRENT down-spell? (F2: drives
	// alerting instead of `prev==""`, so a BORN-down — broken at startup/reseed — alerts too, not
	// just a live up→down transition; re-armed on recovery / reason-change.)
}

// New builds a checker over the probe + sink seams.
func New(probe Probe, sink Sink, logger *log.Logger) *Checker {
	return &Checker{probe: probe, sink: sink, logger: logger}
}

// Check runs one probe cycle: classify, debounce, reflect on the dashboard, and notify on a real
// transition. Best-effort + idempotent — safe to call on a timer. Never returns an error to the
// scheduler (a probe failure IS the signal, not a job failure).
func (c *Checker) Check(ctx context.Context) error {
	pctx, cancel := context.WithTimeout(ctx, probeTimeout)
	constructionErr, perr := c.probe(pctx)
	cancel()

	c.mu.Lock()
	defer c.mu.Unlock()

	// Channel UP.
	if perr == nil {
		c.consecutiveDown = 0
		c.sink.SetDashboard(false, "", "")
		prev := c.state
		c.state = "up"
		c.alerted = false // re-arm for the next down-spell
		if prev != "" && prev != "up" {
			c.logger.Printf("[INFO] [channel] agent channel recovered (was %s)", prev)
			c.sink.NotifyRecovered()
		}
		return nil // healthy first-obs / steady-up → no notify
	}

	// Channel DOWN — classify + debounce transient reasons.
	cls := classify(constructionErr, perr)
	c.consecutiveDown++
	if cls.debounce && c.consecutiveDown < debounceThreshold {
		// A transient blip (e.g. the ~1s agent-restart socket gap, or the agent not yet up on a cold
		// boot). Hold the previous state — do NOT flip the dashboard or notify. Unseeded → assume up
		// until confirmed (so a transient born-down still needs N>=2 before it alerts).
		if c.state == "" {
			c.state = "up"
		}
		c.logger.Printf("[DEBUG] [channel] transient down (%s, %d/%d) — suppressed pending confirmation: %v",
			cls.reason, c.consecutiveDown, debounceThreshold, perr)
		return nil
	}

	// Confirmed down. F2: a NEW down-spell — coming from up/unseeded OR a reason change — re-arms the
	// alert, so a BORN-down (broken at startup/reseed) alerts on cycle 1 for non-transient reasons,
	// not only a live up->down transition. A steady down that already alerted does not re-fire.
	newState := "down:" + string(cls.reason)
	c.sink.SetDashboard(true, cls.reason, cls.hungarian) // dashboard reflects current state always
	prev := c.state
	if prev == "" || prev == "up" || prev != newState {
		c.alerted = false
	}
	c.state = newState
	if c.alerted {
		return nil // steady down, same reason, already alerted → no duplicate (dashboard stays set)
	}
	c.logger.Printf("[WARN] [channel] agent channel DOWN (%s->%s): %v", orUnseeded(prev), newState, perr)
	c.sink.NotifyDown(cls.reason, cls.eventType, cls.severity, cls.english)
	c.alerted = true
	return nil
}

func orUnseeded(s string) string {
	if s == "" {
		return "unseeded"
	}
	return s
}

// State returns the current channel state (for tests/diagnostics).
func (c *Checker) State() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.state == "" {
		return "unseeded"
	}
	return c.state
}