mesh/merge-gate pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus step; every machine composes with the change as it…
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery delivered
mesh/delivery-group group fix/314-a-large-answer-is-paged-not-lost delivered: every member is delivered
The client library refuses to send a reply over the bus's max_payload, and the controller only logged it: conditions and status answered nobody for hours on 2026-10-08 while calls said each was answered in 130 ms, and the operator's channel read nothing. An answer too large is now held under its call and paged to the caller that asks, on the same subject; a caller that does not page is told in words, and calls says it. The overviews no longer carry every finding: conditions and status list each condition with its newest evidence, doctor at most twenty findings a probe (probe= gives one whole), and the JSON overviews are sent once, as data, instead of twice.
480 lines
16 KiB
Go
480 lines
16 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"os"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// What is wrong, kept until observation says it is not (novox/hq to-be 45 §2, ADR 0227).
|
|
//
|
|
// **`status` used to be the only place the mesh said it was wrong, and only to whoever asked.** Every
|
|
// core failure of research 031 was found by a person looking. A condition is the mesh saying it: raised
|
|
// by a watchdog when a signal is late (signals.go), by a probe when an invariant does not hold
|
|
// (doctor.go), or by an event a provider sends (standing.go); kept on the bus with since-when and
|
|
// evidence; said on the bus as it changes, for the operator's channel to carry; and cleared when an
|
|
// observation says it is resolved. Nobody resolves one by hand. A person who knows silences it, for a
|
|
// while, with a reason, and that is recorded as a hand act.
|
|
|
|
// conditionsFrom is the serving controller's keeper; nil in any other process, which opens its own.
|
|
var conditionsFrom *conditions.Keeper
|
|
|
|
// keeperOn is a keeper over the store on a connection, saying its transitions on that connection.
|
|
func keeperOn(ctx context.Context, conn *nats.Conn) (*conditions.Keeper, error) {
|
|
store, history, err := conditions.OnTheBus(ctx, conn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
js, err := conn.JetStream()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return conditions.NewKeeper(ctx, conditions.Options{Store: store, History: history,
|
|
Teller: link.OverNATS{Conn: conn, JS: js},
|
|
Say: func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) },
|
|
// What status leads with changed: composed again soon (a nudge outside the serving controller
|
|
// does nothing).
|
|
Changed: statusFrom.nudge,
|
|
// Written under the lease, carrying its epoch (novox/hq to-be 45 §6).
|
|
Epoch: func() (uint64, error) { return theLease.epoch(context.WithoutCancel(ctx)) }}), nil
|
|
}
|
|
|
|
// withKeeper runs f with the serving controller's keeper, or one of its own that says everything
|
|
// it was given before it returns.
|
|
func withKeeper(ctx context.Context, f func(*conditions.Keeper) error) error {
|
|
if conditionsFrom != nil {
|
|
return f(conditionsFrom)
|
|
}
|
|
return onTheBus(func(conn *nats.Conn) error {
|
|
k, err := keeperOn(ctx, conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
flushing, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
|
defer cancel()
|
|
k.Close(flushing)
|
|
}()
|
|
return f(k)
|
|
})
|
|
}
|
|
|
|
// openConditions is every open condition, for `status` and `node show`: from the serving keeper, or
|
|
// read from the bus. Where there is no bus to read it from, it says so — never "none open".
|
|
func openConditions(ctx context.Context) ([]conditions.Condition, error) {
|
|
if conditionsFrom != nil {
|
|
return conditionsFrom.Open(ctx)
|
|
}
|
|
if _, err := broker.BusAddress(); err != nil {
|
|
return nil, fmt.Errorf("this process has no bus to read the conditions from: %w", err)
|
|
}
|
|
var out []conditions.Condition
|
|
err := onTheBus(func(conn *nats.Conn) error {
|
|
store, _, err := conditions.OnTheBus(ctx, conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
reading, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
defer cancel()
|
|
out, err = conditions.Read(reading, store)
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
// conditionsUsage is how the verb is typed.
|
|
const conditionsUsage = "conditions [--scope S] [--severity urgent|warning] [--machine M] [--json] | " +
|
|
"conditions show <key> | conditions silence <key> --for <duration> --why <text> | " +
|
|
"conditions history [--days N] [--key K] [--json]"
|
|
|
|
// conditionsCommand is `conditions`, `conditions show`, `conditions silence` and `conditions history`.
|
|
func conditionsCommand(ctx context.Context, args []string) error {
|
|
sub := "list"
|
|
if len(args) > 0 && !strings.HasPrefix(args[0], "-") {
|
|
sub, args = args[0], args[1:]
|
|
}
|
|
switch sub {
|
|
case "list":
|
|
return listConditions(ctx, args)
|
|
case "show":
|
|
return showCondition(ctx, args)
|
|
case "silence":
|
|
return silenceCondition(ctx, args)
|
|
case "history":
|
|
return conditionHistory(ctx, args)
|
|
}
|
|
return errors.New(conditionsUsage)
|
|
}
|
|
|
|
func listConditions(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("conditions", flag.ContinueOnError)
|
|
scope := set.String("scope", "", "only this scope: "+strings.Join(conditions.Scopes, ", "))
|
|
severity := set.String("severity", "", "only urgent, or only warning")
|
|
machine := set.String("machine", "", "only those about this machine")
|
|
asJSON := set.Bool("json", false, "as data")
|
|
if rest, err := parseAround(set, args); err != nil {
|
|
return err
|
|
} else if len(rest) > 0 {
|
|
return errors.New(conditionsUsage)
|
|
}
|
|
if *severity != "" && *severity != string(conditions.Urgent) && *severity != string(conditions.Warning) {
|
|
return fmt.Errorf("a severity is urgent or warning, not %q", *severity)
|
|
}
|
|
open, err := openConditions(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var out []conditions.Condition
|
|
for _, c := range open {
|
|
if (*scope == "" || c.Subject.Scope == *scope) && (*severity == "" || string(c.Severity) == *severity) &&
|
|
(*machine == "" || concerns(c, *machine)) {
|
|
out = append(out, c)
|
|
}
|
|
}
|
|
if *asJSON {
|
|
return printJSON(map[string]any{"conditions": inBrief(out), "open": len(open), "counted": counted(out),
|
|
"note": "urgent first, then oldest first; a condition clears when observation says so, never by hand; " +
|
|
"each with its newest evidence — `conditions key=<key>` gives one whole"})
|
|
}
|
|
if len(out) == 0 {
|
|
if len(open) == 0 {
|
|
fmt.Println("no open conditions")
|
|
} else {
|
|
fmt.Printf("none of the %d open condition(s) is about that\n", len(open))
|
|
}
|
|
return nil
|
|
}
|
|
for _, line := range conditionLines(out, time.Now()) {
|
|
fmt.Println(line)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// briefEvidence is how many observations a condition carries in a list of them: its newest. The store
|
|
// keeps ten, and a list of hundreds of conditions with ten each was the larger half of an answer more
|
|
// than the bus carries in one message (novox/hq issue 314). One condition whole is `conditions key=`.
|
|
const briefEvidence = 1
|
|
|
|
// conditionInBrief is a condition as a list carries it: its newest evidence, with how much more there is
|
|
// and where to read it. Everything else of it, so a reader of the list — the operator's channel among
|
|
// them — reads the same condition it always did.
|
|
type conditionInBrief struct {
|
|
conditions.Condition
|
|
// EvidenceKept is how many observations the condition keeps, when the list shows fewer.
|
|
EvidenceKept int `json:"evidence-kept,omitempty"`
|
|
More string `json:"more,omitempty"`
|
|
}
|
|
|
|
// inBrief is a list of conditions in brief; never null.
|
|
func inBrief(list []conditions.Condition) []conditionInBrief {
|
|
out := make([]conditionInBrief, 0, len(list))
|
|
for _, c := range list {
|
|
b := conditionInBrief{Condition: c}
|
|
if len(c.Evidence) > briefEvidence {
|
|
b.EvidenceKept = len(c.Evidence)
|
|
b.Evidence = c.Evidence[:briefEvidence:briefEvidence]
|
|
b.More = c.Show()
|
|
}
|
|
out = append(out, b)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// counted is how many of a list each source raised of each kind — `D14 stalled: 689` — so a list of
|
|
// hundreds says its shape before its lines.
|
|
func counted(list []conditions.Condition) map[string]int {
|
|
out := map[string]int{}
|
|
for _, c := range list {
|
|
out[c.Source+" "+c.Kind]++
|
|
}
|
|
return out
|
|
}
|
|
|
|
// concerns says whether a condition is about a machine: it names it, or its key does.
|
|
func concerns(c conditions.Condition, machine string) bool {
|
|
if c.Subject.Machine == machine || slices.Contains(c.Subject.Also, machine) {
|
|
return true
|
|
}
|
|
for _, part := range strings.Split(c.Key, ".") {
|
|
if part == machine {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// conditionLines is how a list of conditions reads: one line each, its silence under it.
|
|
func conditionLines(list []conditions.Condition, now time.Time) []string {
|
|
var out []string
|
|
for _, c := range list {
|
|
times := ""
|
|
if c.Count > 1 {
|
|
times = fmt.Sprintf(", raised %d times", c.Count)
|
|
}
|
|
out = append(out, fmt.Sprintf(" %-7s %s — %s (since %s%s)", strings.ToUpper(string(c.Severity)),
|
|
c.Key, c.Summary, c.Raised.Local().Format("2006-01-02 15:04"), times))
|
|
if c.SilencedAt(now) {
|
|
out = append(out, fmt.Sprintf(" silenced until %s by %s: %s",
|
|
c.Silenced.Until.Local().Format("2006-01-02 15:04"), c.Silenced.By, c.Silenced.Why))
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func showCondition(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("conditions show", flag.ContinueOnError)
|
|
asJSON := set.Bool("json", false, "as data")
|
|
rest, err := parseAround(set, args)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(rest) != 1 {
|
|
return errors.New("conditions show <key>")
|
|
}
|
|
key := rest[0]
|
|
var c conditions.Condition
|
|
var found bool
|
|
if conditionsFrom != nil {
|
|
c, found, err = conditionsFrom.Get(ctx, key)
|
|
} else {
|
|
err = onTheBus(func(conn *nats.Conn) error {
|
|
store, _, err := conditions.OnTheBus(ctx, conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
c, found, err = conditions.ReadOne(ctx, store, key)
|
|
return err
|
|
})
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !found {
|
|
return fmt.Errorf("no condition %s is open — `conditions` lists those that are, and `conditions "+
|
|
"history --key %s` what became of it", key, key)
|
|
}
|
|
if *asJSON {
|
|
return printJSON(c)
|
|
}
|
|
now := time.Now()
|
|
fmt.Printf("%s %s\n %s\n\n", strings.ToUpper(string(c.Severity)), c.Key, c.Summary)
|
|
fmt.Printf(" kind %s\n about %s %s", c.Kind, c.Subject.Scope, c.Subject.ID)
|
|
if c.Subject.Machine != "" {
|
|
fmt.Printf(", on %s", c.Subject.Machine)
|
|
}
|
|
fmt.Printf("\n raised by %s\n since %s (%s ago), observed %d time(s), last %s ago\n",
|
|
c.Source, c.Raised.Local().Format("2006-01-02 15:04:05"), roughly(now.Sub(c.Raised)), c.Observations,
|
|
now.Sub(c.LastObserved).Round(time.Second))
|
|
if c.Count > 1 {
|
|
fmt.Printf(" raised %d times, each within ten minutes of clearing\n", c.Count)
|
|
}
|
|
fmt.Printf(" resolved by %s\n", resolverWords(c.Resolver))
|
|
if c.Silenced != nil {
|
|
fmt.Printf(" silenced until %s by %s: %s\n", c.Silenced.Until.Local().Format("2006-01-02 15:04"),
|
|
c.Silenced.By, c.Silenced.Why)
|
|
}
|
|
if len(c.Tried) > 0 {
|
|
fmt.Println("\n tried:")
|
|
for _, t := range c.Tried {
|
|
fmt.Printf(" %s %s — %s: %s\n", t.At.Local().Format("2006-01-02 15:04"), orHealer(t.By), t.What, t.Outcome)
|
|
}
|
|
}
|
|
fmt.Println("\n evidence, newest first:")
|
|
for _, e := range c.Evidence {
|
|
fmt.Printf(" %s %s\n", e.At.Local().Format("2006-01-02 15:04:05"), e.Said)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func resolverWords(r string) string {
|
|
switch r {
|
|
case conditions.ResolverSelf:
|
|
return "itself: it clears when observation says it is resolved"
|
|
case conditions.ResolverOperator:
|
|
return "the operator: nothing in the mesh will repair it"
|
|
case conditions.ResolverAgent:
|
|
return "an agent"
|
|
}
|
|
return r
|
|
}
|
|
|
|
// silenceCondition stops a condition's messages for a while (to-be 45 §2). A hand act: recorded with
|
|
// who and why before it is done, its cause the condition's kind unless one is given.
|
|
func silenceCondition(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("conditions silence", flag.ContinueOnError)
|
|
forFlag := set.String("for", "", "how long: 30m, 4h, 2d — at most 7d")
|
|
acts := addHandActFlags(set)
|
|
rest, err := parseAround(set, args)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(rest) != 1 {
|
|
return errors.New("conditions silence <key> --for <duration> --why <text>")
|
|
}
|
|
key := rest[0]
|
|
if err := acts.require("conditions silence"); err != nil {
|
|
return err
|
|
}
|
|
d, err := parseFor(*forFlag)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if d > conditions.MaxSilence {
|
|
return fmt.Errorf("a condition is silenced for at most %s at once; past it, say so again", conditions.MaxSilence)
|
|
}
|
|
return withKeeper(ctx, func(k *conditions.Keeper) error {
|
|
c, found, err := k.Get(ctx, key)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !found {
|
|
return fmt.Errorf("no condition %s is open — `conditions` lists them. Nothing was silenced", key)
|
|
}
|
|
if strings.TrimSpace(*acts.cause) == "" {
|
|
*acts.cause = c.Kind
|
|
}
|
|
*acts.condition = key
|
|
acts.record(ctx, "conditions silence", []string{key, "--for", *forFlag})
|
|
held, err := k.Silence(ctx, key, d, link.Caller(), *acts.why)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
fmt.Printf("%s is silenced until %s: no message is sent for it until then. It is still open, and "+
|
|
"`status` still says it; it clears when observation says it is resolved\n",
|
|
held.Key, held.Silenced.Until.Local().Format("2006-01-02 15:04"))
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// parseFor reads a duration, days included.
|
|
func parseFor(s string) (time.Duration, error) {
|
|
s = strings.TrimSpace(s)
|
|
if s == "" {
|
|
return 0, errors.New("say for how long: --for 30m, 4h or 2d")
|
|
}
|
|
if days, ok := strings.CutSuffix(s, "d"); ok {
|
|
n, err := strconv.Atoi(days)
|
|
if err != nil || n <= 0 {
|
|
return 0, fmt.Errorf("%q is not a number of days", s)
|
|
}
|
|
return time.Duration(n) * 24 * time.Hour, nil
|
|
}
|
|
d, err := time.ParseDuration(s)
|
|
if err != nil || d <= 0 {
|
|
return 0, fmt.Errorf("%q is not a duration: 30m, 4h or 2d", s)
|
|
}
|
|
return d, nil
|
|
}
|
|
|
|
func conditionHistory(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("conditions history", flag.ContinueOnError)
|
|
days := set.Int("days", 7, "how many days back, at most 90")
|
|
key := set.String("key", "", "only this condition")
|
|
asJSON := set.Bool("json", false, "as data")
|
|
if rest, err := parseAround(set, args); err != nil {
|
|
return err
|
|
} else if len(rest) > 0 {
|
|
return errors.New("conditions history [--days N] [--key K] [--json]")
|
|
}
|
|
since := time.Now().Add(-time.Duration(*days) * 24 * time.Hour)
|
|
var events []conditions.Event
|
|
read := func(h conditions.History) error {
|
|
var err error
|
|
events, err = h.Since(ctx, since)
|
|
return err
|
|
}
|
|
var err error
|
|
if conditionsFrom != nil {
|
|
events, err = conditionsFrom.HistorySince(ctx, since)
|
|
} else {
|
|
err = onTheBus(func(conn *nats.Conn) error {
|
|
_, history, err := conditions.OnTheBus(ctx, conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return read(history)
|
|
})
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var out []conditions.Event
|
|
for _, e := range events {
|
|
if *key == "" || e.Key == *key {
|
|
out = append(out, e)
|
|
}
|
|
}
|
|
if *asJSON {
|
|
if out == nil {
|
|
out = []conditions.Event{}
|
|
}
|
|
return printJSON(map[string]any{"history": out, "days": *days})
|
|
}
|
|
if len(out) == 0 {
|
|
fmt.Printf("nothing was raised, changed or cleared in the last %d day(s)\n", *days)
|
|
return nil
|
|
}
|
|
for _, e := range out {
|
|
line := fmt.Sprintf("%s %-13s %s", e.At.Local().Format("2006-01-02 15:04:05"), e.Change, e.Key)
|
|
switch e.Change {
|
|
case conditions.ChangeRaised, conditions.ChangeReopened:
|
|
line += " — " + e.Summary
|
|
case conditions.ChangeSeverity:
|
|
line += fmt.Sprintf(" — %s, was %s", e.Severity, e.Was)
|
|
case conditions.ChangeResolver:
|
|
line += fmt.Sprintf(" — %s, was %s", e.Resolver, e.Was)
|
|
default:
|
|
if e.Why != "" {
|
|
line += " — " + e.Why
|
|
}
|
|
}
|
|
fmt.Println(line)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// printConditions is the status section that leads it: every open condition, urgent first, oldest
|
|
// first, silenced ones with their expiry (to-be 45 §2). A store that could not be read is said, and
|
|
// is not "none open".
|
|
func printConditions(list []conditions.Condition, unread string, now time.Time) {
|
|
if unread != "" {
|
|
fmt.Printf("the open conditions could NOT be read, so whether anything is wrong is not known: %s\n\n", unread)
|
|
return
|
|
}
|
|
if len(list) == 0 {
|
|
return
|
|
}
|
|
urgent := 0
|
|
for _, c := range list {
|
|
if c.Severity == conditions.Urgent {
|
|
urgent++
|
|
}
|
|
}
|
|
fmt.Printf("%d open condition(s), %d urgent:\n\n", len(list), urgent)
|
|
for _, line := range conditionLines(list, now) {
|
|
fmt.Println(line)
|
|
}
|
|
fmt.Printf("\n `conditions show <key>` says more; each clears when observation says it is resolved, " +
|
|
"never by hand — `conditions silence <key> --for <d> --why <text>` stops its messages\n\n")
|
|
}
|
|
|
|
// orHealer is who tried, as an attempt names it.
|
|
func orHealer(by string) string {
|
|
if by == "" {
|
|
return "a healer"
|
|
}
|
|
return by
|
|
}
|