mesh-tools
013dad4688
Swaps three import paths for mesh-sdk's go/catalogue, go/overlay and go/validate (the same packages, copied from this repository at 6c2a8d62) and lets gofmt order the groups. The six swap commits compile together once the pin commit (go.mod, modules.txt) is on the branch; each alone does not, because a file on the SDK's types cannot call one still on the old package.
254 lines
10 KiB
Go
254 lines
10 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// A delivery held past its bound, said by the controller and healed by H2 from mesh-delivery's own table
|
|
// (novox/hq ADR 0239 decision 9, to-be 47 Phase B). The conditions are the controller's, one owner: the
|
|
// self-check reads the delivery owner's `stalled` (probe D14) and raises `delivery.<id>.stalled`; healer
|
|
// H2 calls the owner's `close`, which takes only the transition the table names for that state, and only
|
|
// the next observation says whether it worked (ADR 0231).
|
|
|
|
// probeDeliveriesID is the probe that reads the delivery's owner.
|
|
const probeDeliveriesID = "D14"
|
|
|
|
// kindDeliveryStalled is what a delivery held past its bound raises: H2's kind, in the delivery scope.
|
|
const kindDeliveryStalled = "stalled"
|
|
|
|
// deliveryOwnerWithin is how long the owner is given to answer.
|
|
var deliveryOwnerWithin = 10 * time.Second
|
|
|
|
// stalledLine is one delivery past its bound, as mesh-delivery's `stalled` says it.
|
|
type stalledLine struct {
|
|
ID string `json:"id"`
|
|
// Number is the pull request's, for a line of a head the forge never announced (novox/hq issue 347).
|
|
Number int `json:"number,omitempty"`
|
|
State string `json:"state"`
|
|
// Walk is the walk a line is about: a waiting walk no live delivery holds, or, asked with `walks`, each walk
|
|
// waiting for the owner's word and what holds it (novox/hq issues 362, 529).
|
|
Walk string `json:"walk,omitempty"`
|
|
// Waiting are the merge checks a line of the state `unanswered` waits on, as mesh-delivery says each:
|
|
// "mesh/merge-gate pending since <UTC time>", "mesh/repo-check never set" (novox/hq issue 438).
|
|
Waiting []string `json:"waiting,omitempty"`
|
|
// Checks are the same merge checks as data, which the controller words in the operator's own time (novox/hq
|
|
// issue 443): a time inside a finished sentence cannot be said again in another zone. A mesh-delivery from before
|
|
// says none, and Waiting is said as it reads.
|
|
Checks []waitingCheck `json:"checks,omitempty"`
|
|
// Held is the state of the delivery of a line of the state `unanswered`, ready or rejected, when one is held
|
|
// (novox/hq issues 441, 451): its check answered once, so a recheck is the step. Empty for a head with no
|
|
// delivery, and from a mesh-delivery that does not say it yet; the words are never read for it.
|
|
Held string `json:"held,omitempty"`
|
|
For string `json:"for"`
|
|
Bound string `json:"bound"`
|
|
H2 string `json:"h2"`
|
|
Says string `json:"says"`
|
|
// Severity is "urgent" for a group held or half merged and a refused member (novox/hq issues 1275, 1278); empty
|
|
// or anything else is a warning, as before.
|
|
Severity string `json:"severity,omitempty"`
|
|
}
|
|
|
|
// severity is the condition's severity a stalled line asks: urgent when it says so, a warning otherwise.
|
|
func (l stalledLine) severity() conditions.Severity {
|
|
if l.Severity == string(conditions.Urgent) {
|
|
return conditions.Urgent
|
|
}
|
|
return conditions.Warning
|
|
}
|
|
|
|
// waitingCheck is one merge check a stalled line waits on, as mesh-delivery gives it: its context, its state —
|
|
// "pending", or "never-set" — and, for a pending one, since when in RFC 3339, empty when the forge did not say.
|
|
type waitingCheck struct {
|
|
Context string `json:"context"`
|
|
State string `json:"state"`
|
|
Since string `json:"since,omitempty"`
|
|
}
|
|
|
|
// wordsNow is the time the words are said at, which says whether a time needs its day; a seam a test replaces.
|
|
var wordsNow = time.Now
|
|
|
|
// wordsZone is the zone a stalled line's times are said in: the controller's local zone, as every other time in its
|
|
// messages. A seam a test replaces, so that no test writes time.Local, which every goroutine of the package reads.
|
|
var wordsZone = func() *time.Location { return time.Local }
|
|
|
|
// operatorsOnly is whether the table leaves H2 nothing to do for the line: the state is the operator's.
|
|
func (l stalledLine) operatorsOnly() bool { return l.H2 == "" || strings.HasPrefix(l.H2, "none") }
|
|
|
|
// askDeliveryOwner asks the holder of the mesh-delivery seat one of the verbs the controller is granted;
|
|
// a seam a test replaces. The holder's own refusal is an error naming it.
|
|
var askDeliveryOwner = func(ctx context.Context, conn *nats.Conn, verb string, args map[string]any) (json.RawMessage, error) {
|
|
granted := false
|
|
for _, v := range broker.VerbsTheControllerAsksTheDeliveryOwner {
|
|
granted = granted || v.Verb == verb
|
|
}
|
|
if !granted {
|
|
return nil, fmt.Errorf("the controller asks %s.%s, which its grant does not name", catalogue.DeliverySeat, verb)
|
|
}
|
|
if conn == nil {
|
|
return nil, errors.New("this controller is not on the bus")
|
|
}
|
|
answer, err := link.AskMeshSeatTool(ctx, conn, catalogue.DeliverySeat, verb, args, deliveryOwnerWithin)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if answer.Error != "" {
|
|
return nil, fmt.Errorf("%s.%s refused: %s", catalogue.DeliverySeat, verb, answer.Error)
|
|
}
|
|
return unwrapToolResult(answer.Result), nil
|
|
}
|
|
|
|
// unwrapToolResult is a tool's answer whatever the runtime wrapped it in: the JSON itself, or the
|
|
// protocol's content list holding it as text.
|
|
func unwrapToolResult(raw json.RawMessage) json.RawMessage {
|
|
var wrapped struct {
|
|
Content []struct {
|
|
Text string `json:"text"`
|
|
} `json:"content"`
|
|
}
|
|
if json.Unmarshal(raw, &wrapped) == nil && len(wrapped.Content) > 0 && json.Valid([]byte(wrapped.Content[0].Text)) {
|
|
return json.RawMessage(wrapped.Content[0].Text)
|
|
}
|
|
return raw
|
|
}
|
|
|
|
// deliveriesStalled is what the owner says is held past its bound; nothing when no holder is on record,
|
|
// or none answers (D3 says that one).
|
|
func deliveriesStalled(ctx context.Context, conn *nats.Conn, held bool) ([]stalledLine, error) {
|
|
if !held {
|
|
return nil, nil
|
|
}
|
|
raw, err := askDeliveryOwner(ctx, conn, "stalled", map[string]any{})
|
|
if errors.Is(err, link.ErrNothingServes) {
|
|
return nil, nil // the holder not running is D3's holder-silent, said once there
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var lines []stalledLine
|
|
if err := json.Unmarshal(raw, &lines); err != nil {
|
|
return nil, fmt.Errorf("%s.stalled answered something unreadable: %w", catalogue.DeliverySeat, err)
|
|
}
|
|
return lines, nil
|
|
}
|
|
|
|
// walkWaits are, by walk, the lines mesh-delivery says of a walk waiting for its word: what holds it.
|
|
type walkWaits map[string][]stalledLine
|
|
|
|
// walksWaiting asks the delivery's owner what holds each walk waiting for its word (novox/hq issue 529): its
|
|
// `stalled` asked with `walks`, so a line of `plans` names a held delivery, its reason and the verb that releases
|
|
// it, instead of pointing at a hand start. Nil when no holder is on record or none answers: the line then reads as
|
|
// before. An owner from before this answers its stalled lines, of which only those naming a walk are kept.
|
|
func walksWaiting(ctx context.Context, conn *nats.Conn, held bool) walkWaits {
|
|
if !held {
|
|
return nil
|
|
}
|
|
raw, err := askDeliveryOwner(ctx, conn, "stalled", map[string]any{"walks": true})
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
var lines []stalledLine
|
|
if json.Unmarshal(raw, &lines) != nil {
|
|
return nil
|
|
}
|
|
out := walkWaits{}
|
|
for _, l := range lines {
|
|
if l.Walk != "" {
|
|
out[l.Walk] = append(out[l.Walk], l)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// stalledObservations are the conditions of what is stalled.
|
|
func stalledObservations(lines []stalledLine) []conditions.Observation {
|
|
out := make([]conditions.Observation, 0, len(lines))
|
|
for _, l := range lines {
|
|
o := conditions.Observation{Scope: conditions.ScopeDelivery, ID: l.ID, Kind: kindDeliveryStalled,
|
|
Severity: l.severity(),
|
|
Summary: fmt.Sprintf("the delivery %s has been %s for %s, past its bound of %s (%s): healer H2 may %s — "+
|
|
"`mesh-delivery.show %s`", l.ID, l.State, l.For, l.Bound, l.Says, l.H2, l.ID),
|
|
Said: fmt.Sprintf("%s for %s", l.State, l.For)}
|
|
if l.operatorsOnly() {
|
|
o.Resolver = conditions.ResolverOperator
|
|
}
|
|
o.Headline, o.Explanation, o.Resolved, o.Needs, o.Actions = stalledWords(l, o)
|
|
out = append(out, o)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// probeDeliveries is D14: every delivery held past its state's bound is said.
|
|
func probeDeliveries(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
|
|
entries, err := d.open.inventory.Catalogued(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var conn *nats.Conn
|
|
if d.js != nil {
|
|
conn = d.js.Conn()
|
|
}
|
|
lines, err := deliveriesStalled(ctx, conn, deliverySeatHeld(entries))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return stalledObservations(lines), nil
|
|
}
|
|
|
|
// --- H2, for a delivery ---------------------------------------------------------------------------------
|
|
|
|
// connOf is the bus a healer asks over.
|
|
func (h *healing) connOf() *nats.Conn {
|
|
if h.js == nil {
|
|
return nil
|
|
}
|
|
return h.js.Conn()
|
|
}
|
|
|
|
// appliesToAStalledDelivery is H2's for a delivery: the owner still lists it, with a transition the table
|
|
// lets H2 take; never one whose state is the operator's.
|
|
func appliesToAStalledDelivery(ctx context.Context, h *healing, c conditions.Condition) (string, bool, string, error) {
|
|
lines, err := deliveriesStalled(ctx, h.connOf(), true)
|
|
if err != nil {
|
|
return "", false, "", err
|
|
}
|
|
for _, l := range lines {
|
|
if l.ID != c.Subject.ID {
|
|
continue
|
|
}
|
|
if l.operatorsOnly() {
|
|
return "", false, "the delivery is " + l.State + ", a state the table leaves to the operator", nil
|
|
}
|
|
return c.Key, true, "", nil
|
|
}
|
|
return "", false, "the delivery's owner no longer lists it as stalled: its condition clears on the next look", nil
|
|
}
|
|
|
|
// repairDelivery is H2 for a delivery: the owner's `close`, which reads the walk again and takes only the
|
|
// transition its table names. A refusal is no repair, said; the next observation says whether it worked.
|
|
func repairDelivery(ctx context.Context, h *healing, c conditions.Condition) (string, string, error) {
|
|
raw, err := askDeliveryOwner(ctx, h.connOf(), "close", map[string]any{"id": c.Subject.ID, "why": c.Key})
|
|
if err != nil {
|
|
if errors.Is(err, link.ErrNothingServes) {
|
|
return "", "", err
|
|
}
|
|
return "closed nothing", oneLine(err.Error()), nil
|
|
}
|
|
var said string
|
|
if json.Unmarshal(raw, &said) != nil {
|
|
said = string(raw)
|
|
}
|
|
return "asked " + catalogue.DeliverySeat + " to close " + c.Subject.ID, said, nil
|
|
}
|