package main import ( "context" "encoding/json" "errors" "fmt" "strings" "time" "github.com/nats-io/nats.go" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/catalogue" "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..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 ", "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"` } // 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: conditions.Warning, 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 }