jschoubben
87410d79ed
mesh/merge-gate head 87410d79edd6 merged into trunk e231185572ba — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 87410d79edd6 merged into trunk e231185572ba — pass: its merge-check.sh passed
mesh/delivery delivered
225 lines
8.5 KiB
Go
225 lines
8.5 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// A channel's verified sender is composed into a membership only while the machines are judged root-free at
|
|
// that moment (novox/hq ADR 0259 §8), and the judgement fails closed when a read fails (novox/hq issue 930):
|
|
// a push composed during a failed read issued the router a membership without it, and nothing composed it
|
|
// again, so every approval was refused until somebody pushed. Two things follow.
|
|
//
|
|
// What a composition drops it says (droppedVerifiedSenders), with the reason the judgement gave.
|
|
//
|
|
// And the self-check issues it again (reissueVerifiedSenders): each time it judges the machines, a membership
|
|
// that lacks a verified sender the composition would now give it is issued again with only its kinds changed.
|
|
// **Bound: one self-check, doctorEvery (five minutes), after the judgement passes.** It only ever adds: a
|
|
// machine that stops passing keeps what it was issued, as at a push, because the router asks the root-free
|
|
// verb again before it honours an approval, and that is the check that holds.
|
|
|
|
// droppedVerifiedSenders says, one line each, every kind that declares `verified-sender` and was composed
|
|
// without it, naming the kind, its machine and why. whys is each machine's verdict reason (rootVerdict.Why).
|
|
func droppedVerifiedSenders(r broker.Records, where broker.Placements, whys map[string]string) []string {
|
|
var routerNodes []string
|
|
for node, declared := range r.Assigned {
|
|
for _, d := range declared {
|
|
for _, s := range d.Holds {
|
|
if s.Name == routerSeat && !slices.Contains(routerNodes, node) {
|
|
routerNodes = append(routerNodes, node)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
sort.Strings(routerNodes)
|
|
var out []string
|
|
for _, k := range where.Kinds {
|
|
declared := false
|
|
for _, d := range r.Assigned[k.Node] {
|
|
for _, s := range d.Holds {
|
|
if d.Module == k.Module && s.Name == k.Seat && s.Kind == k.Kind && slices.Contains(s.Capabilities, "verified-sender") {
|
|
declared = true
|
|
}
|
|
}
|
|
}
|
|
if !declared || slices.Contains(k.Capabilities, "verified-sender") {
|
|
continue
|
|
}
|
|
var why string
|
|
switch {
|
|
case !r.RootFree[k.Node]:
|
|
why = k.Node + " is not judged root-free: " + whys[k.Node]
|
|
case len(routerNodes) != 1:
|
|
why = fmt.Sprintf("the router is placed on %d machines, not one", len(routerNodes))
|
|
case !r.RootFree[routerNodes[0]]:
|
|
why = "the router's machine " + routerNodes[0] + " is not judged root-free: " + whys[routerNodes[0]]
|
|
default:
|
|
why = k.Module + " does not run as an account of its own"
|
|
}
|
|
out = append(out, fmt.Sprintf(" %s kind %s of %s on %s is composed without its verified sender: %s",
|
|
k.Seat, k.Kind, k.Module, k.Node, why))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// withVerifiedSenders is a membership as issued with its seat traffic's kinds replaced by the composed ones,
|
|
// and whether that changed it: only when a composed kind carries a verified sender the issued one lacks. Pure.
|
|
func withVerifiedSenders(issued []byte, composed *broker.SeatTraffic) ([]byte, bool, error) {
|
|
if composed == nil {
|
|
return issued, false, nil
|
|
}
|
|
var fields map[string]json.RawMessage
|
|
if err := json.Unmarshal(issued, &fields); err != nil {
|
|
return nil, false, err
|
|
}
|
|
var traffic map[string]json.RawMessage
|
|
var had []broker.KindHeld
|
|
if raw, ok := fields["seat-traffic"]; ok {
|
|
if err := json.Unmarshal(raw, &traffic); err != nil {
|
|
return nil, false, err
|
|
}
|
|
if raw, ok := traffic["kinds"]; ok {
|
|
if err := json.Unmarshal(raw, &had); err != nil {
|
|
return nil, false, err
|
|
}
|
|
}
|
|
}
|
|
gains := false
|
|
for _, k := range composed.Kinds {
|
|
if !slices.Contains(k.Capabilities, "verified-sender") {
|
|
continue
|
|
}
|
|
held := false
|
|
for _, h := range had {
|
|
if h.Seat == k.Seat && h.Kind == k.Kind && h.Module == k.Module && h.Node == k.Node &&
|
|
slices.Contains(h.Capabilities, "verified-sender") {
|
|
held = true
|
|
}
|
|
}
|
|
gains = gains || !held
|
|
}
|
|
if !gains {
|
|
return issued, false, nil
|
|
}
|
|
if traffic == nil {
|
|
traffic = map[string]json.RawMessage{}
|
|
}
|
|
raw, err := json.Marshal(composed.Kinds)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
traffic["kinds"] = raw
|
|
if fields["seat-traffic"], err = json.Marshal(traffic); err != nil {
|
|
return nil, false, err
|
|
}
|
|
body, err := json.Marshal(fields)
|
|
return body, true, err
|
|
}
|
|
|
|
// reissueVerifiedSenders issues again each membership that lacks a verified sender the records now give it:
|
|
// read as last issued, its kinds replaced, published only if no newer one was issued since (a push runs beside
|
|
// the self-check, and its newer membership must not be overwritten by the older fields read here). One never issued is left to its own send. Failures are
|
|
// said and passed over; the next self-check tries again.
|
|
func reissueVerifiedSenders(r broker.Records, where broker.Placements,
|
|
last func(node, module string) ([]byte, uint64, error), publish func(node, module string, body []byte, seq uint64) error,
|
|
say func(format string, args ...any)) int {
|
|
issued := 0
|
|
for node, declared := range r.Assigned {
|
|
for _, d := range declared {
|
|
composed := broker.MembershipFor(node, d, where).SeatTraffic
|
|
if composed == nil || !slices.ContainsFunc(composed.Kinds, func(k broker.KindHeld) bool {
|
|
return slices.Contains(k.Capabilities, "verified-sender")
|
|
}) {
|
|
continue
|
|
}
|
|
was, seq, err := last(node, d.Module)
|
|
if err != nil {
|
|
if !errors.Is(err, nats.ErrMsgNotFound) {
|
|
say(" %s on %s: its membership could not be read to give its verified sender back: %v", d.Module, node, err)
|
|
}
|
|
continue
|
|
}
|
|
body, changed, err := withVerifiedSenders(was, composed)
|
|
if err != nil {
|
|
say(" %s on %s: its membership could not be read to give its verified sender back: %v", d.Module, node, err)
|
|
continue
|
|
}
|
|
if !changed {
|
|
continue
|
|
}
|
|
if err := publish(node, d.Module, body, seq); err != nil {
|
|
// A push issued another since it was read: that one is newer, and the next self-check looks again.
|
|
if strings.Contains(err.Error(), "wrong last sequence") {
|
|
say(" %s on %s: a newer membership was issued meanwhile; its verified sender is looked at again next time",
|
|
d.Module, node)
|
|
continue
|
|
}
|
|
say(" %s on %s: its verified sender could not be issued again: %v", d.Module, node, err)
|
|
continue
|
|
}
|
|
say(" %s on %s: issued its membership again, with the verified sender its machines now pass for", d.Module, node)
|
|
issued++
|
|
}
|
|
}
|
|
return issued
|
|
}
|
|
|
|
// rootVerdictsOf is each machine's verdict reason, for saying why one was not free.
|
|
func rootVerdictsOf(verdicts []rootVerdict) (free map[string]bool, whys map[string]string) {
|
|
free, whys = map[string]bool{}, map[string]string{}
|
|
for _, v := range verdicts {
|
|
if v.Free {
|
|
free[v.Machine] = true
|
|
}
|
|
whys[v.Machine] = strings.TrimSpace(v.Why)
|
|
}
|
|
return free, whys
|
|
}
|
|
|
|
// giveVerifiedSendersBack is the self-check's step: every machine judged, the memberships that lost a verified
|
|
// sender to a failed judgement issued again. Never fails the probe: what could not be done is said.
|
|
func giveVerifiedSendersBack(ctx context.Context, d *doctor) {
|
|
if d.js == nil || d.open == nil || d.open.inventory == nil {
|
|
return
|
|
}
|
|
records, err := d.open.inventory.BusRecords(ctx)
|
|
if err != nil {
|
|
fmt.Printf(" the memberships could not be read to give a verified sender back: %v\n", err)
|
|
return
|
|
}
|
|
// Judged for every machine the records name, as a push does, not only those the self-check judged.
|
|
r := newRootReader(ctx, d.open.inventory, d.js.Conn())
|
|
all := judgeRootFree(ctx, r, records.Nodes, time.Now())
|
|
records.RootFree, _ = rootVerdictsOf(all)
|
|
where := broker.PlacementsOf(records, records.Interchangeable)
|
|
reissueVerifiedSendersOn(ctx, link.OverNATS{Conn: d.js.Conn(), JS: d.js.Context()}, records, where)
|
|
}
|
|
|
|
// reissueVerifiedSendersOn is reissueVerifiedSenders on the bus: memberships read from the AssignmentsStream and
|
|
// published to it.
|
|
func reissueVerifiedSendersOn(ctx context.Context, bus link.OverNATS, records broker.Records, where broker.Placements) int {
|
|
return reissueVerifiedSenders(records, where,
|
|
func(node, module string) ([]byte, uint64, error) {
|
|
m, err := bus.JS.GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject(node, module))
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
return m.Data, m.Sequence, nil
|
|
},
|
|
func(node, module string, body []byte, seq uint64) error {
|
|
return bus.PublishMembershipAfter(ctx, node, module, body, seq)
|
|
},
|
|
func(format string, args ...any) { fmt.Printf(format+"\n", args...) })
|
|
}
|