From b2e2bf0638791c466ba0a7efa17de45f1f4a787e Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 16 Oct 2026 17:42:23 +0200 Subject: [PATCH 1/3] 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 | 17 +- cmd/mesh-controller/push.go | 8 +- .../reissue_verified_sender.go | 210 ++++++++++++++++++ .../reissue_verified_sender_test.go | 108 +++++++++ 4 files changed, 330 insertions(+), 13 deletions(-) create mode 100644 cmd/mesh-controller/reissue_verified_sender.go create mode 100644 cmd/mesh-controller/reissue_verified_sender_test.go diff --git a/cmd/mesh-controller/probe_agent_root.go b/cmd/mesh-controller/probe_agent_root.go index 95111959..847d0d83 100644 --- a/cmd/mesh-controller/probe_agent_root.go +++ b/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 -} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index bc47d401..be0d5b52 100644 --- a/cmd/mesh-controller/push.go +++ b/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 diff --git a/cmd/mesh-controller/reissue_verified_sender.go b/cmd/mesh-controller/reissue_verified_sender.go new file mode 100644 index 00000000..c50492a0 --- /dev/null +++ b/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...) }) +} diff --git a/cmd/mesh-controller/reissue_verified_sender_test.go b/cmd/mesh-controller/reissue_verified_sender_test.go new file mode 100644 index 00000000..eb04ce8e --- /dev/null +++ b/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) + } +} From 1c3c3edc7d48f7bf74c22afd8b6a28e04f753022 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 16 Oct 2026 17:43:25 +0200 Subject: [PATCH 2/3] Prove on a real bus a lost verified sender is issued again once the judgement passes (issue 930) --- .../reissue_verified_sender.go | 9 +++- .../reissue_verified_sender_test.go | 50 +++++++++++++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/cmd/mesh-controller/reissue_verified_sender.go b/cmd/mesh-controller/reissue_verified_sender.go index c50492a0..3666f54b 100644 --- a/cmd/mesh-controller/reissue_verified_sender.go +++ b/cmd/mesh-controller/reissue_verified_sender.go @@ -196,8 +196,13 @@ func giveVerifiedSendersBack(ctx context.Context, d *doctor) { 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, + 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, error) { m, err := bus.JS.GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject(node, module)) if err != nil { diff --git a/cmd/mesh-controller/reissue_verified_sender_test.go b/cmd/mesh-controller/reissue_verified_sender_test.go index eb04ce8e..9741d49f 100644 --- a/cmd/mesh-controller/reissue_verified_sender_test.go +++ b/cmd/mesh-controller/reissue_verified_sender_test.go @@ -8,6 +8,8 @@ import ( "github.com/nats-io/nats.go" "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/link" + "github.com/novox/mesh-controller/internal/testbus" ) func verifiedSenderRecords(free map[string]bool) (broker.Records, broker.Declared) { @@ -106,3 +108,51 @@ func TestAMembershipNeverIssuedIsLeftToItsOwnSend(t *testing.T) { t.Fatalf("issued %d, said %v", n, said) } } + +// **On a real bus** (novox/hq issue 930): a router's membership issued without the verified sender because the +// judgement failed is, once the judgement passes, published again to the AssignmentsStream with it in its seat +// traffic's kinds, and a second pass publishes nothing. +func TestNatsAMembershipIssuedWithoutItsVerifiedSenderIsIssuedAgainWhenTheJudgementPasses(t *testing.T) { + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if err := broker.AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + bus := link.OverNATS{Conn: js.Conn(), JS: js.Context()} + + failed, router := verifiedSenderRecords(nil) + body, err := json.Marshal(broker.MembershipFor("novox", router, broker.PlacementsOf(failed, nil))) + if err != nil { + t.Fatal(err) + } + if err := bus.PublishMembership(t.Context(), "novox", "messenger", body); err != nil { + t.Fatal(err) + } + issued := func() []byte { + m, err := js.Context().GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject("novox", "messenger")) + if err != nil { + t.Fatal(err) + } + return m.Data + } + if hasVerified(t, issued()) { + t.Fatal("issued with a failed judgement, and the verified sender is kept: the test proves nothing") + } + if n := reissueVerifiedSendersOn(t.Context(), bus, failed, broker.PlacementsOf(failed, nil)); n != 0 { + t.Fatalf("issued %d memberships while the judgement still fails", n) + } + passing, _ := verifiedSenderRecords(map[string]bool{"novox": true}) + where := broker.PlacementsOf(passing, nil) + if n := reissueVerifiedSendersOn(t.Context(), bus, passing, where); n != 1 { + t.Fatalf("issued %d memberships after the judgement passed, want 1", n) + } + if !hasVerified(t, issued()) { + t.Fatal("the membership on the bus lacks the verified sender after the judgement passed") + } + if n := reissueVerifiedSendersOn(t.Context(), bus, passing, where); n != 0 { + t.Fatalf("a second pass issued %d memberships", n) + } +} From 87410d79edd6ece1459383b6ab9236ab754843d3 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 16 Oct 2026 17:46:18 +0200 Subject: [PATCH 3/3] Issue a verified sender again only over the membership that was read, so a newer push is not overwritten (issue 930) --- .../reissue_verified_sender.go | 25 +++++++---- .../reissue_verified_sender_test.go | 41 ++++++++++++++++--- internal/link/bus.go | 13 ++++++ 3 files changed, 65 insertions(+), 14 deletions(-) diff --git a/cmd/mesh-controller/reissue_verified_sender.go b/cmd/mesh-controller/reissue_verified_sender.go index 3666f54b..0ad71848 100644 --- a/cmd/mesh-controller/reissue_verified_sender.go +++ b/cmd/mesh-controller/reissue_verified_sender.go @@ -128,10 +128,11 @@ func withVerifiedSenders(issued []byte, composed *broker.SeatTraffic) ([]byte, b } // 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 +// 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, error), publish func(node, module string, body []byte) error, + 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 { @@ -142,7 +143,7 @@ func reissueVerifiedSenders(r broker.Records, where broker.Placements, }) { continue } - was, err := last(node, d.Module) + 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) @@ -157,7 +158,13 @@ func reissueVerifiedSenders(r broker.Records, where broker.Placements, if !changed { continue } - if err := publish(node, d.Module, body); err != nil { + 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 } @@ -203,13 +210,15 @@ func giveVerifiedSendersBack(ctx context.Context, d *doctor) { // 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, error) { + func(node, module string) ([]byte, uint64, error) { m, err := bus.JS.GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject(node, module)) if err != nil { - return nil, err + return nil, 0, err } - return m.Data, nil + return m.Data, m.Sequence, nil + }, + func(node, module string, body []byte, seq uint64) error { + return bus.PublishMembershipAfter(ctx, node, module, body, seq) }, - func(node, module string, body []byte) error { return bus.PublishMembership(ctx, node, module, body) }, func(format string, args ...any) { fmt.Printf(format+"\n", args...) }) } diff --git a/cmd/mesh-controller/reissue_verified_sender_test.go b/cmd/mesh-controller/reissue_verified_sender_test.go index 9741d49f..0f00fdaf 100644 --- a/cmd/mesh-controller/reissue_verified_sender_test.go +++ b/cmd/mesh-controller/reissue_verified_sender_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "errors" "strings" "testing" @@ -25,17 +26,28 @@ func verifiedSenderRecords(free map[string]bool) (broker.Records, broker.Declare type fakeBus struct { last map[string][]byte published map[string][]byte + seq map[string]uint64 + // between runs once, after a read and before the write: a push issuing meanwhile. + between func() } -func (f *fakeBus) get(node, module string) ([]byte, error) { +func (f *fakeBus) get(node, module string) ([]byte, uint64, error) { if b, ok := f.last[node+"/"+module]; ok { - return b, nil + return b, f.seq[node+"/"+module], nil } - return nil, nats.ErrMsgNotFound + return nil, 0, nats.ErrMsgNotFound } -func (f *fakeBus) put(node, module string, body []byte) error { +func (f *fakeBus) put(node, module string, body []byte, seq uint64) error { + if f.between != nil { + f.between() + f.between = nil + } + if seq != f.seq[node+"/"+module] { + return errors.New("nats: wrong last sequence") + } f.published[node+"/"+module] = body f.last[node+"/"+module] = body + f.seq[node+"/"+module]++ return nil } @@ -59,7 +71,7 @@ func hasVerified(t *testing.T, membership []byte) bool { func TestAMembershipThatLostItsVerifiedSenderGetsItBackWhenTheJudgementPasses(t *testing.T) { failed, router := verifiedSenderRecords(nil) whereFailed := broker.PlacementsOf(failed, nil) - bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}} + bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}} body, _ := json.Marshal(broker.MembershipFor("novox", router, whereFailed)) bus.last["novox/messenger"] = body if hasVerified(t, body) { @@ -100,7 +112,7 @@ func TestADroppedVerifiedSenderIsSaidWithItsReason(t *testing.T) { func TestAMembershipNeverIssuedIsLeftToItsOwnSend(t *testing.T) { records, _ := verifiedSenderRecords(map[string]bool{"novox": true}) - bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}} + bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}} var said []string n := reissueVerifiedSenders(records, broker.PlacementsOf(records, nil), bus.get, bus.put, func(f string, a ...any) { said = append(said, f) }) @@ -156,3 +168,20 @@ func TestNatsAMembershipIssuedWithoutItsVerifiedSenderIsIssuedAgainWhenTheJudgem t.Fatalf("a second pass issued %d memberships", n) } } + +// A push issuing a newer membership between the self-check's read and its write is not overwritten (issue 930). +func TestANewerMembershipIssuedMeanwhileIsNotOverwritten(t *testing.T) { + failed, router := verifiedSenderRecords(nil) + bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}} + old, _ := json.Marshal(broker.MembershipFor("novox", router, broker.PlacementsOf(failed, nil))) + bus.last["novox/messenger"], bus.seq["novox/messenger"] = old, 1 + newer := []byte(`{"node":"novox","module":"messenger","newer":true}`) + bus.between = func() { bus.last["novox/messenger"], bus.seq["novox/messenger"] = newer, 2 } + passing, _ := verifiedSenderRecords(map[string]bool{"novox": true}) + if n := reissueVerifiedSenders(passing, broker.PlacementsOf(passing, nil), bus.get, bus.put, func(string, ...any) {}); n != 0 { + t.Fatalf("issued %d over a newer membership", n) + } + if string(bus.last["novox/messenger"]) != string(newer) || len(bus.published) != 0 { + t.Fatalf("the newer membership was overwritten: %s", bus.last["novox/messenger"]) + } +} diff --git a/internal/link/bus.go b/internal/link/bus.go index 6d94cada..fb32ab03 100644 --- a/internal/link/bus.go +++ b/internal/link/bus.go @@ -179,6 +179,19 @@ func (b OverNATS) PublishMembership(ctx context.Context, node, module string, bo return nil } +// PublishMembershipAfter issues a membership only if the one read at sequence seq is still the newest of its +// subject: a push that issued another since is not overwritten, and the refusal comes back (novox/hq issue 930). +func (b OverNATS) PublishMembershipAfter(ctx context.Context, node, module string, body []byte, seq uint64) error { + ctx, cancel := context.WithTimeout(ctx, MembershipWait) + defer cancel() + _, err := b.JS.Publish(broker.MembershipSubject(node, module), body, nats.Context(ctx), + nats.ExpectLastSequencePerSubject(seq)) + if err != nil { + return fmt.Errorf("issuing %s on %s its membership after sequence %d: %w", module, node, seq, err) + } + return nil +} + func (b OverNATS) PublishDeclaration(ctx context.Context, node string, body []byte) error { _, err := b.JS.Publish(DeclareSubject(node), body, nats.Context(ctx)) if err != nil {