Give a channel's verified sender back when the root-free judgement passes again, and say when it is dropped (issue 930)

A push composed during a failed judgement issued the router a membership without the verified sender and nothing composed it again, so every approval was refused. The self-check now issues it again within one run; a composition says which kind lost it and why.

cmd/mesh-controller/probe_agent_root.go

@@ -320,7 +320,11 @@ func probeAgentRoot(ctx context.Context, d *doctor) ([]conditions.Observation, e
 		quiet, err := searchStillRunning(ctx, inv, node, n.AgentAccount, h, had, now)
 		return err == nil && quiet
 	}
-	return rootObservations(ctx, r, trustedMachines(r.entries), time.Now()), nil
+	obs := rootObservations(ctx, r, trustedMachines(r.entries), time.Now())
+	// A membership that lost its verified sender to a judgement that failed at a push gets it back here, once
+	// the machines pass (issue 930); the bound is one self-check.
+	giveVerifiedSendersBack(ctx, d)
+	return obs, nil
 }
 
 // rootObservations judges the machines and says each that fails.
@@ -365,14 +369,3 @@ func rootFreeAnswer(ctx context.Context, machines string, now time.Time) (any, e
 	}
 	return map[string]any{"machines": judgeRootFree(ctx, newRootReader(ctx, d.open.inventory, conn), names, now)}, nil
 }
-
-// rootFreeNow is the machines judged root-free, for composing a push's memberships: only those that pass.
-func rootFreeNow(ctx context.Context, r *rootReader, machines []string, now time.Time) map[string]bool {
-	free := map[string]bool{}
-	for _, v := range judgeRootFree(ctx, r, machines, now) {
-		if v.Free {
-			free[v.Machine] = true
-		}
-	}
-	return free
-}

cmd/mesh-controller/push.go

@@ -1333,8 +1333,14 @@ func issueMemberships(ctx context.Context, open *stores, server *link.Server, se
 	}
 	// Which machines are root-free now (novox/hq ADR 0259 §8): a channel's verified sender is composed for the
 	// router only from one, beside a router on one. Judged once per push, by the root-free verb's judgement.
-	records.RootFree = rootFreeNow(ctx, newRootReader(ctx, open.inventory, bus.Conn), records.Nodes, time.Now())
+	var whys map[string]string
+	records.RootFree, whys = rootVerdictsOf(judgeRootFree(ctx, newRootReader(ctx, open.inventory, bus.Conn),
+		records.Nodes, time.Now()))
 	where := broker.PlacementsOf(records, records.Interchangeable)
+	// What the judgement kept off is said, with its reason (novox/hq issue 930); the self-check gives it back.
+	for _, line := range droppedVerifiedSenders(records, where, whys) {
+		fmt.Println(line)
+	}
 	// **Every declared state's bucket, before the memberships that name it** (novox/hq ADR 0201). The
 	// raise at start asserts them too, but a module registered and assigned since would otherwise have
 	// its bucket only after the control plane next restarts — found the first time a module declared

cmd/mesh-controller/reissue_verified_sender.go

@@ -0,0 +1,210 @@
+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. 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, error), publish func(node, module string, body []byte) 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, 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); err != nil {
+				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)
+	bus := link.OverNATS{Conn: d.js.Conn(), JS: d.js.Context()}
+	reissueVerifiedSenders(records, where,
+		func(node, module string) ([]byte, error) {
+			m, err := bus.JS.GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject(node, module))
+			if err != nil {
+				return nil, err
+			}
+			return m.Data, nil
+		},
+		func(node, module string, body []byte) error { return bus.PublishMembership(ctx, node, module, body) },
+		func(format string, args ...any) { fmt.Printf(format+"\n", args...) })
+}

cmd/mesh-controller/reissue_verified_sender_test.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/novox/mesh-controller/internal/broker"
+)
+
+func verifiedSenderRecords(free map[string]bool) (broker.Records, broker.Declared) {
+	tg := broker.Seat{Name: "channel", Scope: "mesh", Accepts: []string{"show"}, Kinded: true, Kind: "telegram",
+		DeclaredBy: "messenger", Capabilities: []string{"choice", "verified-sender"}}
+	router := broker.Declared{Module: "messenger", RunsAs: "messenger", Holds: []broker.Seat{{Name: "operator-channel",
+		Scope: "mesh", Accepts: []string{"ask"}}}, Uses: []broker.Seat{{Name: "channel", Scope: "mesh", Kinded: true,
+		DeclaredBy: "messenger"}}}
+	return broker.Records{Nodes: []string{"novox"}, RootFree: free, Assigned: map[string][]broker.Declared{
+		"novox": {router, {Module: "telegram", RunsAs: "telegram", Holds: []broker.Seat{tg}}}}}, router
+}
+
+type fakeBus struct {
+	last      map[string][]byte
+	published map[string][]byte
+}
+
+func (f *fakeBus) get(node, module string) ([]byte, error) {
+	if b, ok := f.last[node+"/"+module]; ok {
+		return b, nil
+	}
+	return nil, nats.ErrMsgNotFound
+}
+func (f *fakeBus) put(node, module string, body []byte) error {
+	f.published[node+"/"+module] = body
+	f.last[node+"/"+module] = body
+	return nil
+}
+
+func hasVerified(t *testing.T, membership []byte) bool {
+	t.Helper()
+	var m broker.Membership
+	if err := json.Unmarshal(membership, &m); err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range m.SeatTraffic.Kinds {
+		if k.Kind == "telegram" {
+			return strings.Contains(strings.Join(k.Capabilities, ","), "verified-sender")
+		}
+	}
+	t.Fatal("telegram not in the router's membership")
+	return false
+}
+
+// A membership composed while the judgement failed is issued again once it passes, with the verified sender
+// (novox/hq issue 930); while it fails, nothing is given back (ADR 0259 §8).
+func TestAMembershipThatLostItsVerifiedSenderGetsItBackWhenTheJudgementPasses(t *testing.T) {
+	failed, router := verifiedSenderRecords(nil)
+	whereFailed := broker.PlacementsOf(failed, nil)
+	bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}}
+	body, _ := json.Marshal(broker.MembershipFor("novox", router, whereFailed))
+	bus.last["novox/messenger"] = body
+	if hasVerified(t, body) {
+		t.Fatal("composed with a failed judgement, and the verified sender is kept")
+	}
+	say := func(string, ...any) {}
+
+	if n := reissueVerifiedSenders(failed, whereFailed, bus.get, bus.put, say); n != 0 || len(bus.published) != 0 {
+		t.Fatalf("issued %d while the judgement still fails", n)
+	}
+	passing, _ := verifiedSenderRecords(map[string]bool{"novox": true})
+	wherePassing := broker.PlacementsOf(passing, nil)
+	if n := reissueVerifiedSenders(passing, wherePassing, bus.get, bus.put, say); n != 1 {
+		t.Fatalf("issued %d memberships after the judgement passed, want 1", n)
+	}
+	if !hasVerified(t, bus.published["novox/messenger"]) {
+		t.Fatal("the membership issued again lacks the verified sender")
+	}
+	if n := reissueVerifiedSenders(passing, wherePassing, bus.get, bus.put, say); n != 0 {
+		t.Fatalf("issued %d again with nothing changed", n)
+	}
+}
+
+// What a composition drops it says, naming the kind, the machine and the judgement's reason.
+func TestADroppedVerifiedSenderIsSaidWithItsReason(t *testing.T) {
+	records, _ := verifiedSenderRecords(nil)
+	where := broker.PlacementsOf(records, nil)
+	lines := droppedVerifiedSenders(records, where, map[string]string{"novox": "not measured: no verb heard"})
+	if len(lines) != 1 || !strings.Contains(lines[0], "telegram") || !strings.Contains(lines[0], "novox") ||
+		!strings.Contains(lines[0], "not measured: no verb heard") {
+		t.Fatalf("said %q", lines)
+	}
+	records, _ = verifiedSenderRecords(map[string]bool{"novox": true})
+	if lines := droppedVerifiedSenders(records, broker.PlacementsOf(records, nil), nil); len(lines) != 0 {
+		t.Fatalf("said %q of a kind that kept its verified sender", lines)
+	}
+}
+
+func TestAMembershipNeverIssuedIsLeftToItsOwnSend(t *testing.T) {
+	records, _ := verifiedSenderRecords(map[string]bool{"novox": true})
+	bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}}
+	var said []string
+	n := reissueVerifiedSenders(records, broker.PlacementsOf(records, nil), bus.get, bus.put,
+		func(f string, a ...any) { said = append(said, f) })
+	if n != 0 || len(said) != 0 {
+		t.Fatalf("issued %d, said %v", n, said)
+	}
+}