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...) }) }