mesh-controller/cmd/mesh-controller/bus_waits.go

package main

import (
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
)

// A send held for the bus's planned step (novox/hq issue 336).
//
// **A wait only a person can end is said to that person at once.** No send replaces the bus but its planned
// step (sendToEach, ADR 0236), and the step is a person's `bus upgrade`. So while a new bus build waits, every
// walk whose tier sends to the bus's machine is refused, and tries again each tick. The refusal was kept only
// as the walk's note; nothing was raised, and the operator found the hold by asking why a walk did not move.
// Row S17 raises it on the first tick after the first refusal, in the bus's scope, for the operator: what
// waits, behind which bus build, since when, and the verb. A walk held only by it is not late, so S3 leaves it
// out, as it leaves out a walk under a paused build seat. It clears once the bus's machine has been sent the
// build the mesh holds — the step was taken, and no send is refused for the bus any more — whatever the walks'
// notes still say. Whether the new bus came up healthy is the step's own condition (probe DB).

// kindBusStepWaiting is S17's kind: bus.<module>.step-waiting.
const kindBusStepWaiting = "bus-step-waiting"

// busFacts is a new bus build waiting for its planned step, and the sends held for it.
type busFacts struct {
	module, to string
	// from is the build each machine of the bus runs; machines those whose bus the step would replace.
	from     map[string]string
	machines []string
	waits    []busWaitFacts
}

// busWaitFacts is one walk whose send was refused because it would replace the bus.
type busWaitFacts struct {
	plan, repository, commit string
	// modules are what its tier sends: what waits.
	modules []string
	// since is the first refusal: as this controller saw it, or, read back, the save that kept the refusal.
	since time.Time
}

// busRefusedFirst is when this controller first saw each walk refused for the bus's step. The walk's note
// keeps the refusal across a restart; this keeps its moment more exactly than the walk's last save.
var busRefusedFirst = &firstSeen{at: map[string]time.Time{}}

type firstSeen struct {
	mu sync.Mutex
	at map[string]time.Time
}

// mark keeps the first moment an id was seen.
func (s *firstSeen) mark(id string, at time.Time) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if _, seen := s.at[id]; !seen {
		s.at[id] = at
	}
}

// of is when an id was first seen; zero when it was not.
func (s *firstSeen) of(id string) time.Time {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.at[id]
}

// keepOnly forgets every id not given: a walk no longer held is not held since then.
func (s *firstSeen) keepOnly(ids map[string]bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for id := range s.at {
		if !ids[id] {
			delete(s.at, id)
		}
	}
}

// refusedForTheBus is whether an error is the refusal of a send for the bus's planned step.
func refusedForTheBus(err error) bool {
	return err != nil && strings.Contains(err.Error(), errBusWaits.Error())
}

// busWaitsOf is the sends held for the bus's step, from the open walks: each walk not waiting for its
// delivery's word whose note keeps a refusal for this very bus build, while the build would still replace
// the bus on a machine. first is when this controller first saw a walk refused, zero when it did not.
func busWaitsOf(plans []inventory.Plan, b busPending, first func(string) time.Time) busFacts {
	f := busFacts{module: b.module, to: b.to, from: b.from}
	for _, n := range b.machines {
		if b.moves(n) {
			f.machines = append(f.machines, n)
		}
	}
	if len(f.machines) == 0 {
		return f
	}
	for _, p := range plans {
		if !p.Open() || p.Waiting() || !strings.Contains(p.Note, errBusWaits.Error()) || !strings.Contains(p.Note, short(b.to)) {
			continue
		}
		since := p.Updated
		if seen := first(p.ID); !seen.IsZero() && (since.IsZero() || seen.Before(since)) {
			since = seen
		}
		var modules []string
		if p.Tier >= 0 && p.Tier < len(p.Tiers) {
			modules = append(modules, p.Tiers[p.Tier]...)
		} else {
			modules = planModules(p)
		}
		sort.Strings(modules)
		f.waits = append(f.waits, busWaitFacts{plan: p.ID, repository: p.Repository, commit: p.Commit, modules: modules,
			since: since})
	}
	sort.Slice(f.waits, func(i, j int) bool { return f.waits[i].since.Before(f.waits[j].since) })
	return f
}

// heldByTheBus is the walks of a bus's facts, by id: what S3 leaves out.
func (b busFacts) heldByTheBus() map[string]bool {
	out := map[string]bool{}
	for _, w := range b.waits {
		out[w.plan] = true
	}
	return out
}

// busUpgradeVerb is the call that ends the wait, as the summary names it: through the mesh MCP server, with
// why, saying whether the new version can be undone ("reversible": "true") or not ("irreversible": "true").
const busUpgradeVerb = "`mesh_call mesh-controller.bus {\"upgrade\": \"true\", \"why\": \"…\", \"reversible\" or \"irreversible\": \"true\"}`"

// watchBusWaits is S17: a send held for the bus's planned step, said at once to the operator.
func watchBusWaits(f *signalFacts) []conditions.Observation {
	b := f.bus
	if len(b.waits) == 0 || len(b.machines) == 0 {
		return nil
	}
	since := b.waits[0].since
	var walks, what []string
	for _, w := range b.waits {
		walks = append(walks, fmt.Sprintf("%s (%s %s, tier: %s)", w.plan, w.repository, short(w.commit),
			strings.Join(w.modules, ", ")))
		what = append(what, deliveryWhat(w.modules, w.repository))
	}
	var from []string
	for _, n := range b.machines {
		from = append(from, short(orNotKnown(b.from[n])))
	}
	machines := namesWords(b.machines, 3)
	in := f.now.Sub(since)
	return []conditions.Observation{{Scope: conditions.ScopeBus, ID: b.module, Token: "step-waiting",
		Kind: kindBusStepWaiting, Severity: conditions.Warning, Resolver: conditions.ResolverOperator,
		Machine: b.machines[0], Also: b.machines[1:],
		Summary: fmt.Sprintf("sends to %s wait for the bus's planned step: a new bus build (%s %s → %s) would replace "+
			"the bus there, which only a person's %s does; waiting since %s: %s", strings.Join(b.machines, ", "),
			b.module, strings.Join(sortedUnique(from), ", "), short(b.to), busUpgradeVerb,
			since.UTC().Format(time.RFC3339), strings.Join(walks, "; ")),
		Said:     fmt.Sprintf("%d walk(s) refused since %s", len(b.waits), since.UTC().Format(time.RFC3339)),
		Headline: clipWords("Sends to "+machines+" wait for a bus upgrade", conditions.HeadlineMax),
		Explanation: clipWords(fmt.Sprintf("A new version of the mesh's message system is built, and only a person "+
			"installs it. Until then nothing else is sent to %s: %s waits, for %s so far.", machines,
			namesWords(sortedUnique(what), 3), humanDuration(in)), conditions.ExplanationMax),
		Needs:    "start the bus upgrade " + FromMeshMCPServer,
		Resolved: "Resolved: the bus upgrade started, and sends go on"}}
}

// sortedUnique is a list sorted, each once.
func sortedUnique(xs []string) []string {
	seen := map[string]bool{}
	var out []string
	for _, x := range xs {
		if !seen[x] {
			seen[x] = true
			out = append(out, x)
		}
	}
	sort.Strings(out)
	return out
}

// clipWords keeps a plain sentence within a bound, at a word.
func clipWords(s string, n int) string {
	if len(s) <= n {
		return s
	}
	cut := strings.LastIndex(s[:n-1], " ")
	if cut <= 0 {
		cut = n - 1
	}
	return s[:cut] + "…"
}