Prove on a real bus a lost verified sender is issued again once the judgement passes (issue 930)

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 {

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