mesh-catalog/modules/memory-pressure/cmd/memory-pressure/guard.go

package main

import (
	"context"
	"fmt"
	"os"
	"os/user"
	"strconv"
	"strings"
	"sync"
	"time"
)

// The module's long-running code (novox/hq ADR 0198): the guard that warns before the machine kills
// something for memory. It replaces the predecessor's `mem-guard`, a user unit that checked used RAM
// every thirty seconds and called notify-send.
//
// What changed, and why:
//
//   - **Three signals, not one.** Used RAM alone misses the two things that decide whether a kill is
//     coming: swap filling (systemd-oomd kills at SwapUsedLimit) and stall pressure (it kills a unit
//     whose pressure stays over its limit for twenty seconds). The guard warns on whichever comes
//     first, and says which.
//   - **Ten seconds, not thirty.** oomd's own window is twenty seconds; a check every thirty can
//     notice only after oomd has acted. Reading two files every ten seconds costs nothing measurable.
//     The kernel's PSI triggers would wake on pressure alone, but not on RAM or swap filling, so the
//     guard reads all three on one timer rather than run two mechanisms.
//   - **The notification names what to close.** It lists the largest units, not only a percentage.
//   - **An event as well as a notification.** `pressure.high` and `pressure.cleared` reach the mesh
//     whether or not anyone is at the desktop — a server has no desktop at all.
//   - **No session needed.** The runtime runs as the operator's account but outside the graphical
//     session, so it has no session bus address. The guard names the account's own bus,
//     /run/user/<uid>/bus, and speaks to the notification service with busctl, the service manager's
//     client: nothing is installed for it, and a machine without a desktop simply has no such bus.

// Thresholds, as constants until settings exist (novox/hq issue 168). Used RAM is the predecessor's,
// raised by it from 90 to 95 after the lower line fired during ordinary work.
const (
	WarnUsedPercent    = 95.0
	ClearUsedPercent   = 88.0
	WarnSwapPercent    = 80.0 // systemd-oomd's SwapUsedLimit is 90 %
	ClearSwapPercent   = 70.0
	WarnPressureAvg10  = 40.0 // "some" avg10; oomd acts at 60 % (80 % for the user manager) over 20 s
	ClearPressureAvg10 = 10.0
	CheckEvery         = 10 * time.Second
	NotificationID     = 9010 // replaces the previous warning rather than stacking
)

// Verdict is what one check of the machine concluded.
type Verdict struct {
	High    bool     `json:"high"`
	Reasons []string `json:"reasons"`
	Clear   bool     `json:"clear"`
}

// Judge applies the thresholds to a snapshot. High is any warning line crossed; Clear is every value
// below its clearing line, which is what re-arms the guard.
func Judge(s Snapshot) Verdict {
	v := Verdict{Reasons: []string{}}
	clear := true
	if s.UsedPercent >= WarnUsedPercent {
		v.Reasons = append(v.Reasons, fmt.Sprintf("RAM %.0f%% used (warns at %.0f%%)", s.UsedPercent, WarnUsedPercent))
	}
	if s.UsedPercent > ClearUsedPercent {
		clear = false
	}
	if s.SwapTotalGiB > 0 {
		if s.SwapPercent >= WarnSwapPercent {
			v.Reasons = append(v.Reasons, fmt.Sprintf("swap %.0f%% used (warns at %.0f%%; systemd-oomd kills at 90%%)", s.SwapPercent, WarnSwapPercent))
		}
		if s.SwapPercent > ClearSwapPercent {
			clear = false
		}
	}
	if s.PressureSome != nil {
		if s.PressureSome.Avg10 >= WarnPressureAvg10 {
			v.Reasons = append(v.Reasons, fmt.Sprintf("tasks stalled on memory %.0f%% of the last 10 s (warns at %.0f%%)", s.PressureSome.Avg10, WarnPressureAvg10))
		}
		if s.PressureSome.Avg10 > ClearPressureAvg10 {
			clear = false
		}
	}
	v.High = len(v.Reasons) > 0
	v.Clear = clear
	return v
}

// Guard is the notifier's state, shared with the tools that report it.
type Guard struct {
	m      *Machine
	now    func() time.Time
	emit   func(string, any) error
	notify func(ctx context.Context, summary, body string) error

	mu       sync.Mutex
	latched  bool
	since    time.Time
	last     *Verdict
	lastAt   time.Time
	notified string
	err      string
	desktop  string
	stopped  string
}

func NewGuard(m *Machine, emit func(string, any) error) *Guard {
	g := &Guard{m: m, now: time.Now, emit: emit}
	g.notify = g.desktopNotify
	return g
}

// Check is one wake-up: read, judge, and on crossing a line warn once; on clearing every line, re-arm.
func (g *Guard) Check(ctx context.Context) {
	s, err := g.m.Snapshot()
	g.mu.Lock()
	if err != nil {
		g.err = err.Error()
		g.mu.Unlock()
		return
	}
	v := Judge(s)
	now := g.now()
	g.last, g.lastAt, g.err = &v, now, ""
	warn := v.High && !g.latched
	cleared := g.latched && v.Clear
	if warn {
		g.latched, g.since = true, now
	}
	if cleared {
		g.latched = false
	}
	since := g.since
	g.mu.Unlock()

	if warn {
		largest := ByUnit(g.m.Processes(), 3)
		var names []string
		for _, u := range largest {
			names = append(names, fmt.Sprintf("%s %.1f GiB", firstNonEmpty(u.Largest, u.Unit), (u.RSSMiB+u.SwapMiB)/1024))
		}
		body := strings.Join(v.Reasons, "; ")
		if len(names) > 0 {
			body += ". Largest: " + strings.Join(names, ", ")
		}
		body += ". systemd-oomd kills the worst unit if it climbs further."
		g.publish("pressure.high", map[string]any{"reasons": v.Reasons, "used_percent": s.UsedPercent,
			"swap_used_percent": s.SwapPercent, "pressure_some": s.PressureSome, "largest": largest})
		err := g.notify(ctx, "Memory is running low", body)
		g.mu.Lock()
		if err != nil {
			g.desktop = "not reached: " + err.Error()
		} else {
			g.desktop, g.notified = "reached", now.Format(time.RFC3339)
		}
		g.mu.Unlock()
	}
	if cleared {
		g.publish("pressure.cleared", map[string]any{"used_percent": s.UsedPercent, "swap_used_percent": s.SwapPercent,
			"lasted_seconds": int(now.Sub(since).Seconds())})
	}
}

func (g *Guard) publish(eventType string, body map[string]any) {
	if g.emit == nil {
		return
	}
	if err := g.emit(eventType, body); err != nil {
		fmt.Fprintf(os.Stderr, "%s not published: %v\n", eventType, err)
	}
}

// Bus is the operator's session bus socket: the account the runtime names, else the one this process
// runs as.
func Bus() (string, error) {
	uid := os.Getuid()
	if name := os.Getenv("MESH_OPERATOR_ACCOUNT"); name != "" {
		u, err := user.Lookup(name)
		if err != nil {
			return "", fmt.Errorf("the operator account %s: %w", name, err)
		}
		if n, err := strconv.Atoi(u.Uid); err == nil {
			if n != uid && uid != 0 {
				return "", fmt.Errorf("this process runs as uid %d and the operator's bus belongs to uid %d", uid, n)
			}
			uid = n
		}
	}
	return fmt.Sprintf("/run/user/%d/bus", uid), nil
}

// desktopNotify sends one critical notification to the operator's session, replacing the previous one.
func (g *Guard) desktopNotify(ctx context.Context, summary, body string) error {
	bus, err := Bus()
	if err != nil {
		return err
	}
	if _, err := os.Stat(bus); err != nil {
		return fmt.Errorf("no session bus at %s (nobody is logged in to a desktop)", bus)
	}
	_, err = g.m.Run(ctx, "env", "DBUS_SESSION_BUS_ADDRESS=unix:path="+bus,
		"busctl", "--user", "call", "org.freedesktop.Notifications", "/org/freedesktop/Notifications",
		"org.freedesktop.Notifications", "Notify", "susssasa{sv}i",
		"memory-pressure", strconv.Itoa(NotificationID), "dialog-warning", summary, body, "0", "1", "urgency", "y", "2", "0")
	return err
}

// Run is the guard's life: a check every CheckEvery.
func (g *Guard) Run(ctx context.Context) {
	defer func() {
		if r := recover(); r != nil {
			g.mu.Lock()
			g.stopped = fmt.Sprintf("the guard stopped on a fault: %v", r)
			g.mu.Unlock()
			fmt.Fprintln(os.Stderr, g.stopped)
		}
	}()
	t := time.NewTicker(CheckEvery)
	defer t.Stop()
	for {
		g.Check(ctx)
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// GuardReport is the guard as the guard tool shows it.
type GuardReport struct {
	Running      bool           `json:"running"`
	Stopped      string         `json:"stopped,omitempty"`
	Warned       bool           `json:"warning_given"`
	WarnedSince  *time.Time     `json:"warning_since,omitempty"`
	LastCheck    *time.Time     `json:"last_check,omitempty"`
	LastVerdict  *Verdict       `json:"last_verdict,omitempty"`
	LastError    string         `json:"last_error,omitempty"`
	Desktop      string         `json:"desktop"`
	LastNotified string         `json:"last_notified,omitempty"`
	Thresholds   map[string]any `json:"thresholds"`
}

func (g *Guard) Report() GuardReport {
	g.mu.Lock()
	defer g.mu.Unlock()
	r := GuardReport{
		Running: g.stopped == "" && !g.lastAt.IsZero(), Stopped: g.stopped, Warned: g.latched, LastCheck: when(g.lastAt),
		LastVerdict: g.last, LastError: g.err, Desktop: g.desktop, LastNotified: g.notified,
		Thresholds: map[string]any{
			"warn_used_percent": WarnUsedPercent, "clear_used_percent": ClearUsedPercent,
			"warn_swap_percent": WarnSwapPercent, "clear_swap_percent": ClearSwapPercent,
			"warn_pressure_some_avg10": WarnPressureAvg10, "clear_pressure_some_avg10": ClearPressureAvg10,
			"check_every": CheckEvery.String(), "set_by": "constants until settings exist (novox/hq issue 168)",
		},
	}
	if g.latched {
		r.WarnedSince = when(g.since)
	}
	if r.Desktop == "" {
		if bus, err := Bus(); err != nil {
			r.Desktop = "not reachable: " + err.Error()
		} else if _, err := os.Stat(bus); err != nil {
			r.Desktop = "no session bus at " + bus
		} else {
			r.Desktop = "reachable at " + bus + " (not yet used)"
		}
	}
	return r
}

func firstNonEmpty(ss ...string) string {
	for _, s := range ss {
		if s != "" {
			return s
		}
	}
	return ""
}

// when is a time for a report: absent rather than the zero time.
func when(t time.Time) *time.Time {
	if t.IsZero() {
		return nil
	}
	return &t
}