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 {