mesh-catalog/modules/power/cmd/power/watcher.go

package main

import (
	"context"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

// The watcher publishes the machine's power states as this module's events (novox/hq ADR 0211 §4).
//
// **`sleeping` goes out before the machine sleeps.** It holds logind's delay lock on sleep and
// shutdown; when logind announces a sleep it publishes, waits for the bus at most AnnounceWithin,
// and only then lets go, so the lock never holds a machine awake for long. Whatever could not be
// published waits in order and goes out as soon as the bus answers again — after waking, a `woke`
// queued behind a `sleeping` that never left.

// Event types, as the module's manifest declares them in `emits`.
const (
	Booted       = "booted"
	Sleeping     = "sleeping"
	Woke         = "woke"
	ShuttingDown = "shutting-down"
	OnMains      = "on-mains"
	OnBattery    = "on-battery"
	BatteryLow   = "battery-low"
)

// AnnounceWithin bounds how long the lock holds a sleep or a shutdown for the bus: logind's own
// InhibitDelayMaxSec is 5 s, and the machine must sleep when the bus is gone.
const AnnounceWithin = 3 * time.Second

// LowBattery is the charge below which battery-low is said, once per discharge.
const LowBattery = 10

// Signal is what logind says before a sleep or a shutdown (Start true) and after a wake (false).
type Signal struct {
	Shutdown bool
	Start    bool
}

// Login1 is logind as the watcher uses it, behind an interface so it is tested without a bus.
type Login1 interface {
	// Inhibit takes a delay lock; closing the file lets it go.
	Inhibit(what, who, why string) (*os.File, error)
	Signals() <-chan Signal
	Close()
}

// Emitter publishes one event and returns once the bus has it.
type Emitter func(eventType string, body any) error

type queued struct {
	Type string
	Body map[string]any
}

// Watcher is the long-running half of the module.
type Watcher struct {
	m      *Machine
	emit   Emitter
	dial   func() (Login1, error)
	now    func() time.Time
	state  string // where the last announced boot id is kept
	poll   time.Duration
	mu     sync.Mutex
	lock   *os.File
	queue  []queued
	sleep  time.Time
	wake   time.Time
	source string
	low    bool
	issue  string
}

// NewWatcher is a watcher for this machine, emitting through emit and reaching logind through dial.
func NewWatcher(m *Machine, emit Emitter, dial func() (Login1, error)) *Watcher {
	home, _ := os.UserHomeDir()
	return &Watcher{m: m, emit: emit, dial: dial, now: time.Now, poll: 10 * time.Second,
		state: filepath.Join(home, ".local", "state", "mesh-power", "announced-boot")}
}

// Snapshot is what power_state and power_check show of the watcher.
type Snapshot struct {
	Inhibiting bool   `json:"inhibiting"`
	Pending    int    `json:"pending"`
	Problem    string `json:"problem,omitempty"`
	LastSleep  string `json:"last_sleep,omitempty"`
	LastWake   string `json:"last_wake,omitempty"`
	Source     string `json:"source,omitempty"`
}

func (w *Watcher) Snapshot() Snapshot {
	w.mu.Lock()
	defer w.mu.Unlock()
	s := Snapshot{Inhibiting: w.lock != nil, Pending: len(w.queue), Problem: w.issue, Source: w.source}
	if !w.sleep.IsZero() {
		s.LastSleep = w.sleep.Format(time.RFC3339)
	}
	if !w.wake.IsZero() {
		s.LastWake = w.wake.Format(time.RFC3339)
	}
	return s
}

func (w *Watcher) problem(s string) {
	w.mu.Lock()
	w.issue = s
	w.mu.Unlock()
}

// enqueue adds an event in order, stamped with when it happened.
func (w *Watcher) enqueue(eventType string, body map[string]any) {
	if body == nil {
		body = map[string]any{}
	}
	body["at"] = w.now().UTC().Format(time.RFC3339)
	w.mu.Lock()
	w.queue = append(w.queue, queued{eventType, body})
	w.mu.Unlock()
}

// flush publishes what waits, in order, and stops at the first the bus does not take.
func (w *Watcher) flush() {
	for {
		w.mu.Lock()
		if len(w.queue) == 0 {
			w.mu.Unlock()
			return
		}
		next := w.queue[0]
		w.mu.Unlock()
		if err := w.emit(next.Type, next.Body); err != nil {
			w.problem("the bus did not take " + next.Type + ": " + err.Error())
			return
		}
		w.mu.Lock()
		w.queue = w.queue[1:]
		if len(w.queue) == 0 && strings.HasPrefix(w.issue, "the bus") {
			w.issue = ""
		}
		w.mu.Unlock()
	}
}

// flushWithin publishes what waits, giving up after d: a sleep must not wait for a bus that is gone.
func (w *Watcher) flushWithin(d time.Duration) {
	done := make(chan struct{})
	go func() { w.flush(); close(done) }()
	select {
	case <-done:
	case <-time.After(d):
	}
}

func (w *Watcher) inhibit(l Login1) {
	f, err := l.Inhibit("sleep:shutdown", "mesh power", "say on the bus that this machine sleeps or stops")
	w.mu.Lock()
	defer w.mu.Unlock()
	if err != nil {
		w.issue = "no delay lock: " + err.Error()
		return
	}
	w.lock = f
	if strings.HasPrefix(w.issue, "no delay lock") || strings.HasPrefix(w.issue, "logind") {
		w.issue = "" // solved: a problem that no longer holds is not said
	}
}

// stillHeld says whether the lock's descriptor is still logind's reference; replaceable in tests.
var stillHeld = func(f *os.File) bool { return IsInhibitor(int(f.Fd())) }

// verify takes the lock again when it is no longer held, and says so: a lock lost silently would
// let the machine sleep without a word on the bus.
func (w *Watcher) verify(l Login1) {
	w.mu.Lock()
	lock := w.lock
	w.mu.Unlock()
	if lock != nil && stillHeld(lock) {
		return
	}
	w.mu.Lock()
	w.lock = nil
	w.mu.Unlock()
	w.inhibit(l)
}

func (w *Watcher) release() {
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.lock != nil {
		w.lock.Close()
		w.lock = nil
	}
}

// bootOnce queues `booted` the first time this boot is seen, remembered across restarts of the
// runtime, so a restart is not mistaken for a boot.
func (w *Watcher) bootOnce() {
	id := w.m.read("/proc/sys/kernel/random/boot_id")
	if id == "" {
		return
	}
	if last, _ := os.ReadFile(w.state); strings.TrimSpace(string(last)) == id {
		return
	}
	body := map[string]any{"boot_id": id}
	if bt, ok := w.m.BootTime(); ok {
		body["booted_at"] = bt.UTC().Format(time.RFC3339)
	}
	w.enqueue(Booted, body)
	if err := os.MkdirAll(filepath.Dir(w.state), 0o755); err == nil {
		_ = os.WriteFile(w.state, []byte(id+"\n"), 0o644)
	}
}

// supply queues on-mains, on-battery and battery-low as the machine's supplies change. The first
// reading sets the baseline and says nothing; a machine with no Mains supply says nothing at all.
func (w *Watcher) supply() {
	supplies := w.m.Supplies()
	src := Source(supplies)
	w.mu.Lock()
	was := w.source
	w.source = src
	w.mu.Unlock()
	if src != "none" && was != "" && was != src {
		w.enqueue(src, nil)
	}
	p, _ := Battery(supplies)
	if p == nil {
		return
	}
	w.mu.Lock()
	low := w.low
	w.mu.Unlock()
	switch {
	case src == "on-battery" && *p <= LowBattery && !low:
		w.mu.Lock()
		w.low = true
		w.mu.Unlock()
		w.enqueue(BatteryLow, map[string]any{"percent": *p})
	case src == "on-mains" || *p > LowBattery+5:
		w.mu.Lock()
		w.low = false
		w.mu.Unlock()
	}
}

// Run watches until ctx ends. Without logind it still says boots and supplies, and tries logind
// again every minute.
func (w *Watcher) Run(ctx context.Context) {
	w.bootOnce()
	w.supply()
	tick := time.NewTicker(w.poll)
	defer tick.Stop()
	var l Login1
	var signals <-chan Signal
	retry := time.NewTimer(0)
	defer retry.Stop()
	for {
		select {
		case <-ctx.Done():
			w.release()
			if l != nil {
				l.Close()
			}
			return
		case <-retry.C:
			got, err := w.dial()
			if err != nil {
				w.problem("logind: " + err.Error())
				retry.Reset(time.Minute)
				continue
			}
			l, signals = got, got.Signals()
			w.inhibit(l)
		case s, open := <-signals:
			if !open {
				w.release()
				l, signals = nil, nil
				retry.Reset(time.Minute)
				continue
			}
			w.handle(l, s)
		case <-tick.C:
			if l != nil {
				w.verify(l)
			}
			w.supply()
			w.flush()
		}
	}
}

func (w *Watcher) handle(l Login1, s Signal) {
	switch {
	case s.Shutdown && s.Start:
		w.enqueue(ShuttingDown, nil)
		w.flushWithin(AnnounceWithin)
		w.release()
	case !s.Shutdown && s.Start:
		w.mu.Lock()
		w.sleep = w.now()
		w.mu.Unlock()
		w.enqueue(Sleeping, nil)
		w.flushWithin(AnnounceWithin)
		w.release()
	case !s.Shutdown && !s.Start:
		w.mu.Lock()
		w.wake = w.now()
		w.mu.Unlock()
		w.enqueue(Woke, nil)
		w.inhibit(l)
		w.supply()
		w.flush()
	}
}