agent v0.27.0: slice 10 P3 — self-heal watchdog reconcile + 4-state intent model
IntentStore (durable-id-keyed: new/enrolled/ejected/decommissioned, OnAbsent replug rule). Watchdog re-mounts only enrolled drives (out-of-band unmount heals; ejected/decommissioned/new left alone) + exp-backoff flapping guard (alert@4, cap@8). guest-attach records enrolled; eject records ejected. Non-hollow tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -3,6 +3,29 @@
|
|||||||
All notable changes to **felhom-agent** are recorded here. Update on every code
|
All notable changes to **felhom-agent** are recorded here. Update on every code
|
||||||
change that gets pushed.
|
change that gets pushed.
|
||||||
|
|
||||||
|
## v0.27.0 — slice 10 P3: self-heal watchdog reconcile + 4-state intent model (2026-06-12)
|
||||||
|
|
||||||
|
The storage watchdog goes from detect-only → detect-and-reconcile: the agent autonomously re-mounts an
|
||||||
|
enrolled external drive that dropped out-of-band (the colleague's Proxmox unmount), gated by a persisted
|
||||||
|
INTENT model so it never auto-adopts an unknown drive or fights an official eject.
|
||||||
|
|
||||||
|
- **`internal/storage/intent.go` — `IntentStore`** — durable, **durable-id-keyed** (UUID/WWN, never
|
||||||
|
sdX/path), atomic-write 4-state model: `new` (not recorded → never auto-mount), `enrolled` (desired
|
||||||
|
mounted → reconcile drift), `ejected` (intentional unmount → leave alone), `decommissioned`
|
||||||
|
(permanent). `OnAbsent` clears `ejected`→`enrolled` so a replug auto-mounts (the replug rule).
|
||||||
|
Records intent ONLY through the official enroll/eject paths — an out-of-band unmount records nothing
|
||||||
|
and is healed. Tests cover the states, persistence, the replug rule, and the reconcile gate.
|
||||||
|
- **`watchdog.go` — intent-gated reconcile + flapping guard (3C)** — the re-mount candidate (device
|
||||||
|
present, not mounted) now fires ONLY for an `enrolled` drive (via `IntentReader`); a present→absent
|
||||||
|
transition (device gone) calls `OnAbsent`. Exponential backoff (`debounce·2^fails`) + an alert after
|
||||||
|
4 failed cycles + a hard stop after 8 (no infinite loop). Failure = "still not present a full backoff
|
||||||
|
window after we dispatched" (a slow async re-mount isn't miscounted). Tests: colleague-unmount→
|
||||||
|
reconciled; ejected/new/decommissioned→left alone; ejected→absent→replug→auto-mount; flapping→caps.
|
||||||
|
- **`internal/localapi`** — `POST /disks/guest-attach` records `enrolled`; `POST /disks/eject` records
|
||||||
|
`ejected` (BEFORE unmount, while the durable-id still resolves) via the new `IntentRecorder`. `main.go`
|
||||||
|
opens one `IntentStore` (`<StateDir>/drive-intents.json`) shared by the watchdog + local API; open
|
||||||
|
failure degrades to ungated legacy remount (logged).
|
||||||
|
|
||||||
## v0.26.0 — slice 10 P2 activation: guest-reboot endpoint (user-triggered drive activation) (2026-06-12)
|
## v0.26.0 — slice 10 P2 activation: guest-reboot endpoint (user-triggered drive activation) (2026-06-12)
|
||||||
|
|
||||||
A drive enrolled into a RUNNING unprivileged guest can't be live-activated (proven: `pct set` won't
|
A drive enrolled into a RUNNING unprivileged guest can't be live-activated (proven: `pct set` won't
|
||||||
|
|||||||
@@ -43,7 +43,7 @@ import (
|
|||||||
|
|
||||||
// version is the agent version. Overridable at build time with
|
// version is the agent version. Overridable at build time with
|
||||||
// -ldflags "-X main.version=<v>"; defaults to the in-repo CHANGELOG version.
|
// -ldflags "-X main.version=<v>"; defaults to the in-repo CHANGELOG version.
|
||||||
var version = "0.26.0"
|
var version = "0.27.0"
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
var (
|
var (
|
||||||
@@ -329,10 +329,27 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int {
|
|||||||
storageTrigger := make(chan struct{}, 1)
|
storageTrigger := make(chan struct{}, 1)
|
||||||
loop.SetTrigger(storageTrigger)
|
loop.SetTrigger(storageTrigger)
|
||||||
remounter := &gateRemounter{gate: gate, ops: hostOps, hostID: cfg.Hub.HostID, logger: logger}
|
remounter := &gateRemounter{gate: gate, ops: hostOps, hostID: cfg.Hub.HostID, logger: logger}
|
||||||
|
// Drive intent store (slice 10 P3 self-heal): persisted, durable-id-keyed enroll/eject/decommission
|
||||||
|
// state. Gates the watchdog's self-heal re-mount to ENROLLED drives, and the local API records
|
||||||
|
// enroll/eject into it. Open failure → ungated legacy remount (logged) so the daemon still runs.
|
||||||
|
var intentReader storage.IntentReader
|
||||||
|
intentStore, ierr := storage.OpenIntentStore(filepath.Join(cfg.LocalAPI.StateDir, "drive-intents.json"))
|
||||||
|
if ierr != nil {
|
||||||
|
logger.Warn("storage: intent store unavailable — self-heal runs UNGATED (legacy remount-any)", "err", ierr)
|
||||||
|
intentStore = nil
|
||||||
|
} else {
|
||||||
|
intentReader = intentStore
|
||||||
|
}
|
||||||
watchdog := storage.NewWatchdog(storage.WatchdogOptions{
|
watchdog := storage.NewWatchdog(storage.WatchdogOptions{
|
||||||
Targets: storage.NewCachingKnownTargets(observer, cfg.Storage.KnownRefresh()),
|
Targets: storage.NewCachingKnownTargets(observer, cfg.Storage.KnownRefresh()),
|
||||||
Liveness: storage.NewHostLiveness(hostReader, 0),
|
Liveness: storage.NewHostLiveness(hostReader, 0),
|
||||||
Remounter: remounter,
|
Remounter: remounter,
|
||||||
|
Intent: intentReader, // P3: only re-mount enrolled drives
|
||||||
|
OnAbsent: func(durableID string) { // P3: clear `ejected` on physical absence (replug rule)
|
||||||
|
if intentStore != nil {
|
||||||
|
_ = intentStore.OnAbsent(durableID)
|
||||||
|
}
|
||||||
|
},
|
||||||
Trigger: func() {
|
Trigger: func() {
|
||||||
select {
|
select {
|
||||||
case storageTrigger <- struct{}{}:
|
case storageTrigger <- struct{}{}:
|
||||||
@@ -381,7 +398,11 @@ func runDaemon(cfg config.Config, logger *slog.Logger) int {
|
|||||||
// host still reports/reconciles). The leaf is generated+persisted once so its pin is stable.
|
// host still reports/reconciles). The leaf is generated+persisted once so its pin is stable.
|
||||||
localServers := 0
|
localServers := 0
|
||||||
var localTokens *localapi.TokenStore
|
var localTokens *localapi.TokenStore
|
||||||
localSrv := buildLocalAPIServer(cfg, px, backupStore, observer, hostOps, gate, collector, logger, &localTokens)
|
var intentRec localapi.IntentRecorder
|
||||||
|
if intentStore != nil { // avoid a typed-nil interface (would defeat the nil check)
|
||||||
|
intentRec = intentStore
|
||||||
|
}
|
||||||
|
localSrv := buildLocalAPIServer(cfg, px, backupStore, observer, hostOps, gate, collector, intentRec, logger, &localTokens)
|
||||||
if localTokens != nil {
|
if localTokens != nil {
|
||||||
defer localTokens.Close()
|
defer localTokens.Close()
|
||||||
}
|
}
|
||||||
@@ -532,7 +553,7 @@ func buildRestoreTestScheduler(cfg config.Config, px *proxmox.Client, engine *re
|
|||||||
// leaf (stable fingerprint). Any failure DISABLES the server (returns nil) WITHOUT crashing the
|
// leaf (stable fingerprint). Any failure DISABLES the server (returns nil) WITHOUT crashing the
|
||||||
// daemon — the host still reports/reconciles; only the controller channel is unavailable until
|
// daemon — the host still reports/reconciles; only the controller channel is unavailable until
|
||||||
// fixed. The opened token store is returned via outTokens so the caller can Close it.
|
// fixed. The opened token store is returned via outTokens so the caller can Close it.
|
||||||
func buildLocalAPIServer(cfg config.Config, px *proxmox.Client, store *backup.Store, observer *storage.Observer, hostOps storage.HostOps, gate *reconcile.Gate, collector *hub.Collector, logger *slog.Logger, outTokens **localapi.TokenStore) *localapi.Server {
|
func buildLocalAPIServer(cfg config.Config, px *proxmox.Client, store *backup.Store, observer *storage.Observer, hostOps storage.HostOps, gate *reconcile.Gate, collector *hub.Collector, intent localapi.IntentRecorder, logger *slog.Logger, outTokens **localapi.TokenStore) *localapi.Server {
|
||||||
if !cfg.LocalAPI.Enabled() {
|
if !cfg.LocalAPI.Enabled() {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -575,6 +596,7 @@ func buildLocalAPIServer(cfg config.Config, px *proxmox.Client, store *backup.St
|
|||||||
DiskGate: storageGateAdapter{gate: gate, hostID: cfg.Hub.HostID},
|
DiskGate: storageGateAdapter{gate: gate, hostID: cfg.Hub.HostID},
|
||||||
Guests2: px,
|
Guests2: px,
|
||||||
GuestAttach: guestBinder, // slice 10 P2: bind enrolled data drives into the guest
|
GuestAttach: guestBinder, // slice 10 P2: bind enrolled data drives into the guest
|
||||||
|
Intent: intent, // slice 10 P3: record enroll/eject intent for self-heal
|
||||||
|
|
||||||
// Host metrics (slice 9): the shared collector serves GET /host/metrics — a fresh host +
|
// Host metrics (slice 9): the shared collector serves GET /host/metrics — a fresh host +
|
||||||
// per-storage view to the customer's monitoring page (reuses the slice-4 collector).
|
// per-storage view to the customer's monitoring page (reuses the slice-4 collector).
|
||||||
|
|||||||
@@ -70,6 +70,14 @@ type GuestAttacher interface {
|
|||||||
RebootGuest(ctx context.Context, vmid int) error
|
RebootGuest(ctx context.Context, vmid int) error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// IntentRecorder persists drive enroll/eject INTENT (slice 10 P3 self-heal), keyed by durable-id, so
|
||||||
|
// the watchdog reconciles only enrolled drives and respects an official eject. Satisfied by
|
||||||
|
// *storage.IntentStore. Optional — when nil, the local API records no intent (self-heal is ungated).
|
||||||
|
type IntentRecorder interface {
|
||||||
|
SetEnrolled(durableID string) error
|
||||||
|
SetEjected(durableID string) error
|
||||||
|
}
|
||||||
|
|
||||||
// ---- handlers ---------------------------------------------------------------------------
|
// ---- handlers ---------------------------------------------------------------------------
|
||||||
|
|
||||||
// DiskInfo is one host drive with its data-bearing flag (for the UI).
|
// DiskInfo is one host drive with its data-bearing flag (for the UI).
|
||||||
@@ -205,6 +213,10 @@ func (s *Server) handleDiskEject(w http.ResponseWriter, r *http.Request, vmid in
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
dependents := s.dependentGuests(r.Context(), req.Where)
|
dependents := s.dependentGuests(r.Context(), req.Where)
|
||||||
|
// Record the EJECT intent BEFORE unmounting (the target still resolves to its durable-id) so the
|
||||||
|
// self-heal watchdog leaves it alone — an OFFICIAL eject is the only thing that sets this (P3); an
|
||||||
|
// out-of-band unmount records nothing and is healed.
|
||||||
|
s.recordIntent(r.Context(), req.Where, "ejected")
|
||||||
if err := s.disks.Unmount(r.Context(), req.Where); err != nil {
|
if err := s.disks.Unmount(r.Context(), req.Where); err != nil {
|
||||||
s.logger.Error("local-api: disk eject", "vmid", vmid, "where", req.Where, "err", err)
|
s.logger.Error("local-api: disk eject", "vmid", vmid, "where", req.Where, "err", err)
|
||||||
writeErr(w, http.StatusBadRequest, "eject failed: "+err.Error())
|
writeErr(w, http.StatusBadRequest, "eject failed: "+err.Error())
|
||||||
@@ -266,6 +278,8 @@ func (s *Server) handleDiskGuestAttach(w http.ResponseWriter, r *http.Request, v
|
|||||||
writeErr(w, http.StatusBadGateway, "guest-attach failed: "+err.Error())
|
writeErr(w, http.StatusBadGateway, "guest-attach failed: "+err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
// Record the drive as ENROLLED so the self-heal watchdog will reconcile it (P3).
|
||||||
|
s.recordIntent(r.Context(), where, "enrolled")
|
||||||
writeOK(w, map[string]any{"vmid": vmid, "attached": where, "slot": slot})
|
writeOK(w, map[string]any{"vmid": vmid, "attached": where, "slot": slot})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -485,6 +499,47 @@ func (s *Server) handleDiskFormat(w http.ResponseWriter, r *http.Request, vmid i
|
|||||||
"device is system/backup-protected — format requires an operator signature ("+dec.Reason+")")
|
"device is system/backup-protected — format requires an operator signature ("+dec.Reason+")")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// durableIDForMount resolves the durable-id of the storage mounted at `where` (from the agent's own
|
||||||
|
// storage view) — the key the intent store records enroll/eject against. "" if not resolvable.
|
||||||
|
func (s *Server) durableIDForMount(ctx context.Context, where string) string {
|
||||||
|
targets, err := s.storage.Observe(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
for _, t := range targets {
|
||||||
|
if t.MountPath == where {
|
||||||
|
return t.DurableID
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
|
||||||
|
// recordIntent records enroll/eject intent for the drive at `where`, best-effort (a nil store, an
|
||||||
|
// unresolved durable-id, or a write error is logged, never fatal — intent is a self-heal aid, not a
|
||||||
|
// gate on the user's action). `action` is "enrolled" or "ejected".
|
||||||
|
func (s *Server) recordIntent(ctx context.Context, where, action string) {
|
||||||
|
if s.intent == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
id := s.durableIDForMount(ctx, where)
|
||||||
|
if id == "" {
|
||||||
|
s.logger.Warn("local-api: intent not recorded — durable-id unresolved", "where", where, "action", action)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
var err error
|
||||||
|
switch action {
|
||||||
|
case "enrolled":
|
||||||
|
err = s.intent.SetEnrolled(id)
|
||||||
|
case "ejected":
|
||||||
|
err = s.intent.SetEjected(id)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
s.logger.Warn("local-api: intent record failed", "where", where, "action", action, "durable_id", id, "err", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
s.logger.Info("local-api: drive intent recorded", "where", where, "action", action, "durable_id", id)
|
||||||
|
}
|
||||||
|
|
||||||
// hostReader returns the injected root-free host topology reader, or the production default. The seam
|
// hostReader returns the injected root-free host topology reader, or the production default. The seam
|
||||||
// keeps the role classification (SystemDisks) testable without touching the real /proc /dev /sys.
|
// keeps the role classification (SystemDisks) testable without touching the real /proc /dev /sys.
|
||||||
func (s *Server) hostReader() storage.HostReader {
|
func (s *Server) hostReader() storage.HostReader {
|
||||||
|
|||||||
@@ -84,6 +84,9 @@ type Options struct {
|
|||||||
// GuestAttach binds an enrolled user-data drive's felhom-data namespace into the guest (slice 10
|
// GuestAttach binds an enrolled user-data drive's felhom-data namespace into the guest (slice 10
|
||||||
// P2, Model A). OPTIONAL — when nil, POST /disks/guest-attach reports "not configured".
|
// P2, Model A). OPTIONAL — when nil, POST /disks/guest-attach reports "not configured".
|
||||||
GuestAttach GuestAttacher
|
GuestAttach GuestAttacher
|
||||||
|
// Intent records drive enroll/eject intent for the self-heal watchdog (slice 10 P3). OPTIONAL —
|
||||||
|
// when nil, no intent is recorded (self-heal runs ungated).
|
||||||
|
Intent IntentRecorder
|
||||||
// HostReader is the root-free host topology reader used to classify a device/mount's protection
|
// HostReader is the root-free host topology reader used to classify a device/mount's protection
|
||||||
// ROLE (it backs SystemDisks for the eject role-gate + the /disks role hints). OPTIONAL — when nil
|
// ROLE (it backs SystemDisks for the eject role-gate + the /disks role hints). OPTIONAL — when nil
|
||||||
// it defaults to the production *storage.ProcHostReader. Injectable so the role-gate is testable.
|
// it defaults to the production *storage.ProcHostReader. Injectable so the role-gate is testable.
|
||||||
@@ -139,6 +142,7 @@ type Server struct {
|
|||||||
diskGate StorageGate // slice 8C (optional)
|
diskGate StorageGate // slice 8C (optional)
|
||||||
guestList GuestLister // slice 8C (optional)
|
guestList GuestLister // slice 8C (optional)
|
||||||
guestAttach GuestAttacher // slice 10 P2 (optional)
|
guestAttach GuestAttacher // slice 10 P2 (optional)
|
||||||
|
intent IntentRecorder // slice 10 P3 (optional)
|
||||||
host storage.HostReader // role classification source (optional; defaults to ProcHostReader)
|
host storage.HostReader // role classification source (optional; defaults to ProcHostReader)
|
||||||
|
|
||||||
hostMetrics HostMetricsProvider // slice 9 (optional)
|
hostMetrics HostMetricsProvider // slice 9 (optional)
|
||||||
@@ -180,6 +184,7 @@ func NewServer(o Options) (*Server, error) {
|
|||||||
diskGate: o.DiskGate,
|
diskGate: o.DiskGate,
|
||||||
guestList: o.Guests2,
|
guestList: o.Guests2,
|
||||||
guestAttach: o.GuestAttach,
|
guestAttach: o.GuestAttach,
|
||||||
|
intent: o.Intent,
|
||||||
host: o.HostReader,
|
host: o.HostReader,
|
||||||
hostMetrics: o.HostMetrics,
|
hostMetrics: o.HostMetrics,
|
||||||
hostID: o.HostID,
|
hostID: o.HostID,
|
||||||
|
|||||||
@@ -0,0 +1,131 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"sync"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Drive INTENT model (slice 10 P3 self-heal). The agent persists, per external user-data drive
|
||||||
|
// (keyed by DURABLE-ID — UUID/WWN, never sdX/path), the operator/customer INTENT — which is distinct
|
||||||
|
// from "is it currently mounted". The self-heal reconciler acts ONLY on this intent, so an
|
||||||
|
// out-of-band unmount (a colleague's Proxmox action) — which records NO intent — gets healed, while an
|
||||||
|
// official eject is respected.
|
||||||
|
//
|
||||||
|
// Four states (3A):
|
||||||
|
// - new : durable-id NOT in the store. Never auto-mounted (the user must enroll it).
|
||||||
|
// - enrolled : desired = mounted + bound into the guest. Drift (present && unmounted) with
|
||||||
|
// this intent → reconcile.
|
||||||
|
// - ejected : an intentional, temporary unmount, set ONLY via the official eject endpoint.
|
||||||
|
// Present && unmounted → left alone. CLEARED to `enrolled` when the drive goes
|
||||||
|
// physically ABSENT, so a replug auto-mounts (the replug rule, for free).
|
||||||
|
// - decommissioned : permanent. Never auto-mounted; SURVIVES absent/present; cleared only by an
|
||||||
|
// explicit re-commission.
|
||||||
|
//
|
||||||
|
// The elegant invariant: intent is recorded ONLY through the official enroll/eject/decommission
|
||||||
|
// paths. "Is it unmounted" ≠ "did someone officially eject it."
|
||||||
|
type DriveIntent string
|
||||||
|
|
||||||
|
const (
|
||||||
|
IntentNew DriveIntent = "" // not in the store
|
||||||
|
IntentEnrolled DriveIntent = "enrolled" // desired mounted → reconcile drift
|
||||||
|
IntentEjected DriveIntent = "ejected" // intentional unmount → leave alone
|
||||||
|
IntentDecommissioned DriveIntent = "decommissioned" // permanent → never auto-mount
|
||||||
|
)
|
||||||
|
|
||||||
|
// IntentStore is the durable, durable-id-keyed intent map. Thread-safe; atomic file writes
|
||||||
|
// (tmp+rename), 0600. Shared between the local-API (records enroll/eject) and the self-heal
|
||||||
|
// reconciler (reads to gate remounts; clears `ejected` on absent).
|
||||||
|
type IntentStore struct {
|
||||||
|
path string
|
||||||
|
mu sync.Mutex
|
||||||
|
m map[string]DriveIntent // durable-id -> intent
|
||||||
|
}
|
||||||
|
|
||||||
|
// OpenIntentStore loads (or initializes) the store at path. A missing file is an empty store; a
|
||||||
|
// corrupt file is an error (fail loud — the reconciler's gate must not silently lose intent).
|
||||||
|
func OpenIntentStore(path string) (*IntentStore, error) {
|
||||||
|
s := &IntentStore{path: path, m: map[string]DriveIntent{}}
|
||||||
|
data, err := os.ReadFile(path)
|
||||||
|
if err != nil {
|
||||||
|
if os.IsNotExist(err) {
|
||||||
|
return s, nil
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("intent store: read %s: %w", path, err)
|
||||||
|
}
|
||||||
|
if len(data) > 0 {
|
||||||
|
if err := json.Unmarshal(data, &s.m); err != nil {
|
||||||
|
return nil, fmt.Errorf("intent store: parse %s: %w", path, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return s, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get returns the intent for a durable-id (IntentNew if absent). An empty durable-id is always
|
||||||
|
// IntentNew — the reconciler must never act on a drive whose identity it can't pin.
|
||||||
|
func (s *IntentStore) Get(durableID string) DriveIntent {
|
||||||
|
if durableID == "" {
|
||||||
|
return IntentNew
|
||||||
|
}
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
return s.m[durableID]
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetEnrolled records a drive as enrolled (on successful enroll/guest-attach). Idempotent.
|
||||||
|
func (s *IntentStore) SetEnrolled(durableID string) error { return s.set(durableID, IntentEnrolled) }
|
||||||
|
|
||||||
|
// SetEjected records an intentional eject (official eject endpoint only).
|
||||||
|
func (s *IntentStore) SetEjected(durableID string) error { return s.set(durableID, IntentEjected) }
|
||||||
|
|
||||||
|
// SetDecommissioned records a permanent decommission (operator path).
|
||||||
|
func (s *IntentStore) SetDecommissioned(durableID string) error {
|
||||||
|
return s.set(durableID, IntentDecommissioned)
|
||||||
|
}
|
||||||
|
|
||||||
|
// OnAbsent transitions a drive that has gone physically ABSENT: an `ejected` drive becomes `enrolled`
|
||||||
|
// again (so a replug auto-mounts — the replug rule); `decommissioned` and `enrolled` are unchanged;
|
||||||
|
// `new` stays new. This is the ONLY place ejected→enrolled happens.
|
||||||
|
func (s *IntentStore) OnAbsent(durableID string) error {
|
||||||
|
if durableID == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
if s.m[durableID] == IntentEjected {
|
||||||
|
s.m[durableID] = IntentEnrolled
|
||||||
|
return s.saveLocked()
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *IntentStore) set(durableID string, intent DriveIntent) error {
|
||||||
|
if durableID == "" {
|
||||||
|
return fmt.Errorf("intent store: refusing to record intent for an empty durable-id")
|
||||||
|
}
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
if s.m[durableID] == intent {
|
||||||
|
return nil // idempotent — no write
|
||||||
|
}
|
||||||
|
s.m[durableID] = intent
|
||||||
|
return s.saveLocked()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *IntentStore) saveLocked() error {
|
||||||
|
data, err := json.MarshalIndent(s.m, "", " ")
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
tmp := s.path + ".tmp"
|
||||||
|
if err := os.WriteFile(tmp, data, 0o600); err != nil {
|
||||||
|
os.Remove(tmp)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return os.Rename(tmp, s.path)
|
||||||
|
}
|
||||||
@@ -0,0 +1,91 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestIntentStore_StatesAndPersistence(t *testing.T) {
|
||||||
|
path := filepath.Join(t.TempDir(), "intent.json")
|
||||||
|
s, err := OpenIntentStore(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("open: %v", err)
|
||||||
|
}
|
||||||
|
// new: absent → IntentNew
|
||||||
|
if s.Get("uuid:A") != IntentNew {
|
||||||
|
t.Fatalf("absent should be new, got %q", s.Get("uuid:A"))
|
||||||
|
}
|
||||||
|
if err := s.SetEnrolled("uuid:A"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := s.SetEjected("uuid:B"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if err := s.SetDecommissioned("uuid:C"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
// reload → states persisted
|
||||||
|
s2, err := OpenIntentStore(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("reopen: %v", err)
|
||||||
|
}
|
||||||
|
if s2.Get("uuid:A") != IntentEnrolled || s2.Get("uuid:B") != IntentEjected || s2.Get("uuid:C") != IntentDecommissioned {
|
||||||
|
t.Fatalf("states not persisted: A=%q B=%q C=%q", s2.Get("uuid:A"), s2.Get("uuid:B"), s2.Get("uuid:C"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestIntentStore_OnAbsent_ReplyRule(t *testing.T) {
|
||||||
|
s, _ := OpenIntentStore(filepath.Join(t.TempDir(), "i.json"))
|
||||||
|
_ = s.SetEjected("uuid:E")
|
||||||
|
_ = s.SetDecommissioned("uuid:D")
|
||||||
|
_ = s.SetEnrolled("uuid:N")
|
||||||
|
|
||||||
|
// ejected → absent → enrolled (so a replug auto-mounts)
|
||||||
|
if err := s.OnAbsent("uuid:E"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if s.Get("uuid:E") != IntentEnrolled {
|
||||||
|
t.Fatalf("ejected→absent should clear to enrolled, got %q", s.Get("uuid:E"))
|
||||||
|
}
|
||||||
|
// decommissioned survives absent
|
||||||
|
_ = s.OnAbsent("uuid:D")
|
||||||
|
if s.Get("uuid:D") != IntentDecommissioned {
|
||||||
|
t.Fatalf("decommissioned must survive absent, got %q", s.Get("uuid:D"))
|
||||||
|
}
|
||||||
|
// enrolled unchanged by absent
|
||||||
|
_ = s.OnAbsent("uuid:N")
|
||||||
|
if s.Get("uuid:N") != IntentEnrolled {
|
||||||
|
t.Fatalf("enrolled should stay enrolled, got %q", s.Get("uuid:N"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestIntentStore_EmptyDurableIDRefused(t *testing.T) {
|
||||||
|
s, _ := OpenIntentStore(filepath.Join(t.TempDir(), "i.json"))
|
||||||
|
if err := s.SetEnrolled(""); err == nil {
|
||||||
|
t.Fatal("recording intent for an empty durable-id must be refused")
|
||||||
|
}
|
||||||
|
if s.Get("") != IntentNew {
|
||||||
|
t.Fatal("empty durable-id must read as new")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ReconcilePermitted is the gate the reconciler uses: only an `enrolled` drive (intent-none) is
|
||||||
|
// auto-mounted; new/ejected/decommissioned are not. This documents the decision table.
|
||||||
|
func TestIntentStore_ReconcileGate(t *testing.T) {
|
||||||
|
s, _ := OpenIntentStore(filepath.Join(t.TempDir(), "i.json"))
|
||||||
|
_ = s.SetEnrolled("uuid:enr")
|
||||||
|
_ = s.SetEjected("uuid:ej")
|
||||||
|
_ = s.SetDecommissioned("uuid:dec")
|
||||||
|
cases := map[string]bool{
|
||||||
|
"uuid:enr": true, // enrolled → reconcile
|
||||||
|
"uuid:ej": false, // ejected → leave alone
|
||||||
|
"uuid:dec": false, // decommissioned → never
|
||||||
|
"uuid:new": false, // new → never auto-mount
|
||||||
|
}
|
||||||
|
for id, want := range cases {
|
||||||
|
got := s.Get(id) == IntentEnrolled
|
||||||
|
if got != want {
|
||||||
|
t.Errorf("reconcile-permitted(%s): got %v want %v (intent %q)", id, got, want, s.Get(id))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+138
-32
@@ -63,6 +63,22 @@ type Transition struct {
|
|||||||
To string
|
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
|
// 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
|
// 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.
|
// an immediate, debounced out-of-band host-report so the hub learns of a drop in seconds.
|
||||||
@@ -80,13 +96,19 @@ type Watchdog struct {
|
|||||||
now func() time.Time
|
now func() time.Time
|
||||||
spawn func(func()) // spawn a background task (overridable in tests; default `go f()`)
|
spawn func(func()) // spawn a background task (overridable in tests; default `go f()`)
|
||||||
|
|
||||||
mu sync.Mutex
|
intent IntentReader // P3: gate re-mount by drive intent (nil = ungated legacy)
|
||||||
last map[string]bool // name -> last observed present (only for seen targets)
|
onAbsent func(durableID string) // P3: called when a known target goes ABSENT (clears `ejected`)
|
||||||
lastFire time.Time
|
|
||||||
fired bool // lastFire is valid
|
mu sync.Mutex
|
||||||
pending bool // a transition is awaiting the debounce window
|
last map[string]bool // name -> last observed present (only for seen targets)
|
||||||
lastRemount map[string]time.Time // name -> last re-mount dispatch (rate-limit)
|
lastFire time.Time
|
||||||
lastUUID map[string]string // name -> last fs-UUID observed while ATTACHED (re-mount key)
|
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
|
// WatchdogOptions configures a Watchdog. Targets, Liveness and Trigger are required; the
|
||||||
@@ -99,6 +121,11 @@ type WatchdogOptions struct {
|
|||||||
Interval time.Duration
|
Interval time.Duration
|
||||||
Debounce time.Duration
|
Debounce time.Duration
|
||||||
Logger *slog.Logger
|
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
|
// NewWatchdog builds a Watchdog. A nil Trigger is tolerated (the watchdog still tracks
|
||||||
@@ -121,21 +148,49 @@ func NewWatchdog(opts WatchdogOptions) *Watchdog {
|
|||||||
trigger = func() {}
|
trigger = func() {}
|
||||||
}
|
}
|
||||||
return &Watchdog{
|
return &Watchdog{
|
||||||
targets: opts.Targets,
|
targets: opts.Targets,
|
||||||
liveness: opts.Liveness,
|
liveness: opts.Liveness,
|
||||||
remounter: opts.Remounter,
|
remounter: opts.Remounter,
|
||||||
interval: interval,
|
intent: opts.Intent,
|
||||||
debounce: debounce,
|
onAbsent: opts.OnAbsent,
|
||||||
trigger: trigger,
|
interval: interval,
|
||||||
logger: logger,
|
debounce: debounce,
|
||||||
now: func() time.Time { return time.Now().UTC() },
|
trigger: trigger,
|
||||||
spawn: func(f func()) { go f() },
|
logger: logger,
|
||||||
last: map[string]bool{},
|
now: func() time.Time { return time.Now().UTC() },
|
||||||
lastRemount: map[string]time.Time{},
|
spawn: func(f func()) { go f() },
|
||||||
lastUUID: map[string]string{},
|
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
|
// Run fast-polls until ctx is cancelled. The first tick establishes the baseline (no
|
||||||
// trigger); subsequent ticks detect transitions. Returns nil on ctx cancellation.
|
// trigger); subsequent ticks detect transitions. Returns nil on ctx cancellation.
|
||||||
func (w *Watchdog) Run(ctx context.Context) error {
|
func (w *Watchdog) Run(ctx context.Context) error {
|
||||||
@@ -207,24 +262,52 @@ func (w *Watchdog) tick(ctx context.Context) {
|
|||||||
var transitions []Transition
|
var transitions []Transition
|
||||||
var remounts []KnownTarget
|
var remounts []KnownTarget
|
||||||
current := make(map[string]bool, len(probes))
|
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 {
|
for _, p := range probes {
|
||||||
current[p.t.Name] = p.present
|
current[p.t.Name] = p.present
|
||||||
if prev, seen := w.last[p.t.Name]; seen && prev != 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)})
|
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
|
||||||
// Re-mount candidate: a mount-backed target that is NOT mounted but whose backing
|
// !present AND whose device is no longer present. The intent store clears `ejected` here so
|
||||||
// device is physically present (a disconnected→device-back state). Rate-limited per
|
// a replug auto-mounts (the replug rule); flapping state resets (it's gone, not flapping).
|
||||||
// target to the debounce window so a persistent mount failure can't storm HostOps.
|
if prev && !p.present && p.t.MountBacked && !p.devicePresent {
|
||||||
if w.remounter != nil && p.t.MountBacked && !p.present && p.devicePresent {
|
absentNow = append(absentNow, p.t)
|
||||||
if last, ok := w.lastRemount[p.t.Name]; !ok || now.Sub(last) >= w.debounce {
|
w.resetFlap(p.t.Name)
|
||||||
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-mount candidate (self-heal): an ENROLLED, mount-backed target that is NOT mounted but
|
||||||
// re-mounts promptly.
|
// 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 {
|
if p.present {
|
||||||
delete(w.lastRemount, p.t.Name)
|
w.resetFlap(p.t.Name)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
w.last = current // targets no longer known drop out
|
w.last = current // targets no longer known drop out
|
||||||
@@ -254,10 +337,33 @@ func (w *Watchdog) tick(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
for _, t := range remounts {
|
for _, t := range remounts {
|
||||||
t := t
|
t := t
|
||||||
w.logger.Info("storage: watchdog dispatching benign re-mount (device returned)",
|
w.logger.Info("storage: watchdog self-heal — re-mounting enrolled drive (device returned, no eject intent)",
|
||||||
"target", t.Name, "where", t.MountPath)
|
"target", t.Name, "where", t.MountPath, "durable_id", t.DurableID)
|
||||||
w.spawn(func() { w.remounter.Remount(ctx, t) })
|
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 {
|
func stateStr(present bool) string {
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"net"
|
"net"
|
||||||
|
"path/filepath"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -358,3 +359,124 @@ func TestHostLiveness_NetworkDial(t *testing.T) {
|
|||||||
t.Error("network target without endpoint must not be flagged down")
|
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())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user