From 3f038b989e32d7cfa6b559b5312b49285263f953 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 11:59:21 +0200 Subject: [PATCH] Hear a merge's announcement from whichever module claims the delivery seat (issue 1211) --- .../first_builds_integration_test.go | 4 +- cmd/mesh-controller/merged_by_claim_test.go | 37 +++++++ cmd/mesh-controller/moved.go | 18 +++- cmd/mesh-controller/moved_bus_test.go | 4 +- cmd/mesh-controller/new_module_bus_test.go | 4 +- cmd/mesh-controller/own_path_by_claim_test.go | 7 ++ cmd/mesh-controller/push.go | 2 +- internal/broker/streams.go | 14 ++- internal/link/delivery_merged_test.go | 102 ++++++++++++++++++ internal/link/receive_nats.go | 20 +++- internal/link/serve.go | 18 +++- 11 files changed, 216 insertions(+), 14 deletions(-) create mode 100644 cmd/mesh-controller/merged_by_claim_test.go create mode 100644 internal/link/delivery_merged_test.go diff --git a/cmd/mesh-controller/first_builds_integration_test.go b/cmd/mesh-controller/first_builds_integration_test.go index e6f363ce..133c79a7 100644 --- a/cmd/mesh-controller/first_builds_integration_test.go +++ b/cmd/mesh-controller/first_builds_integration_test.go @@ -94,7 +94,7 @@ func TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegis following := following{open: open} server := link.ConnectNats(js, nil, noListener{}) - if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, kept: following.resumeFirstBuilds}); err != nil { + if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: theDeliveryClaims, kept: following.resumeFirstBuilds}); err != nil { t.Fatal(err) } if err := broker.AssertServingConsumers(js); err != nil { @@ -110,7 +110,7 @@ func TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegis word := func() { t.Helper() body, _ := json.Marshal(a.merge) - if _, err := js.Context().Publish(broker.DeliveryMergedSubject, body); err != nil { + if _, err := js.Context().Publish(broker.DeliveryMergedSubjectOf(theDelivery), body); err != nil { t.Fatal(err) } becomes(t, 15*time.Second, "the word being kept", func() bool { diff --git a/cmd/mesh-controller/merged_by_claim_test.go b/cmd/mesh-controller/merged_by_claim_test.go new file mode 100644 index 00000000..1b3544c7 --- /dev/null +++ b/cmd/mesh-controller/merged_by_claim_test.go @@ -0,0 +1,37 @@ +package main + +import ( + "context" + "errors" + "testing" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" +) + +// 1211/AC1 to AC3: the controller keeps a merge announced by a module that claims the delivery seat, whatever its +// name, and takes no word from one that does not (novox/hq ADR 0437 §3). With no bus held, a word that passes the +// claim is held for the bus (ErrTryAgain); a word that does not is refused before the bus is asked. +func TestOnlyAModuleClaimingTheSeatIsTheDeliverysWordOnAMerge(t *testing.T) { + entries := []inventory.Entry{claimingTheSeat("mesh-delivery-gate"), namedOnly("gitea"), namedOnly("mesh-delivery")} + keeper := keepingDeliveryMerges{ + js: func() *broker.JetStream { return nil }, + catalog: func(context.Context) ([]inventory.Entry, error) { return entries, nil }, + } + body := []byte(`{"owner":"novox","repo":"r","number":1,"head_sha":"abc","ask_id":"a","option":"approve"}`) + for emitter, passes := range map[string]bool{ + "mesh-delivery-gate": true, // the claimant, under another name + "gitea": false, // claims no seat + "mesh-delivery": false, // named so, claiming no seat here + } { + err := keeper.KeepDeliveryMerge(context.Background(), emitter, body) + if passes != errors.Is(err, link.ErrTryAgain) || (!passes && !errors.Is(err, link.ErrNotTheDeliverysWord)) { + t.Errorf("%s: %v (passes the claim: %v)", emitter, err, passes) + } + } + entries = []inventory.Entry{claimingTheSeat("mesh-delivery")} + if err := keeper.KeepDeliveryMerge(context.Background(), "mesh-delivery", body); !errors.Is(err, link.ErrTryAgain) { + t.Errorf("today's module, claiming the seat: %v", err) + } +} diff --git a/cmd/mesh-controller/moved.go b/cmd/mesh-controller/moved.go index b0a5276e..361582b6 100644 --- a/cmd/mesh-controller/moved.go +++ b/cmd/mesh-controller/moved.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" "strings" @@ -14,6 +15,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/asks" "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/catalogue" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) @@ -115,12 +117,26 @@ func movedOnApprove(d deliveryMerge, ask *asks.Ask, w *asks.Warrant) (string, er // keepingDeliveryMerges keeps each merge mesh-delivery announces in the controller's bucket for them. type keepingDeliveryMerges struct { js func() *broker.JetStream + // catalog is the catalogue, whose modules claiming the delivery seat are the only ones whose word on a merge is + // kept (novox/hq ADR 0437 §3). + catalog func(ctx context.Context) ([]inventory.Entry, error) // kept, when set, is told each merge once it is kept: a first merge waiting for the word asks its builds then // (novox/hq ADR 0432, issue 1146). kept func(ctx context.Context, d deliveryMerge) error } -func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, body []byte) error { +func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, emitter string, body []byte) error { + if k.catalog == nil { + return fmt.Errorf("%w: the catalogue is not read", link.ErrTryAgain) + } + entries, err := k.catalog(ctx) + if err != nil { + return fmt.Errorf("%w: %w", link.ErrTryAgain, err) + } + if !slices.Contains(claimantsOf(entries, catalogue.DeliverySeat), emitter) { + return fmt.Errorf("%w: %q is not among %v", link.ErrNotTheDeliverysWord, emitter, + claimantsOf(entries, catalogue.DeliverySeat)) + } d, err := parseDeliveryMerge(body) if err != nil { return err diff --git a/cmd/mesh-controller/moved_bus_test.go b/cmd/mesh-controller/moved_bus_test.go index d1fad9de..1e999dd2 100644 --- a/cmd/mesh-controller/moved_bus_test.go +++ b/cmd/mesh-controller/moved_bus_test.go @@ -74,7 +74,7 @@ func TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTh } server := link.ConnectNats(js, nil, noListener{}) - if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream}); err != nil { + if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: theDeliveryClaims}); err != nil { t.Fatal(err) } // The events consumer exists, from now on, before the word is published: Receive would make it too late. @@ -90,7 +90,7 @@ func TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTh // mesh-delivery says it merged the pull request, and the controller keeps what it said. body, _ := json.Marshal(a.merge) - if _, err := js.Context().Publish(broker.DeliveryMergedSubject, body); err != nil { + if _, err := js.Context().Publish(broker.DeliveryMergedSubjectOf(theDelivery), body); err != nil { t.Fatal(err) } becomes(t, 15*time.Second, "the merge being kept", func() bool { diff --git a/cmd/mesh-controller/new_module_bus_test.go b/cmd/mesh-controller/new_module_bus_test.go index c2e88074..e4579d37 100644 --- a/cmd/mesh-controller/new_module_bus_test.go +++ b/cmd/mesh-controller/new_module_bus_test.go @@ -75,7 +75,7 @@ func TestANewModuleFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheB } server := link.ConnectNats(js, nil, noListener{}) - if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream}); err != nil { + if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: theDeliveryClaims}); err != nil { t.Fatal(err) } // The events consumer exists, from now on, before the word is published: Receive would make it too late. @@ -91,7 +91,7 @@ func TestANewModuleFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheB // mesh-delivery says it merged the pull request, and the controller keeps what it said. body, _ := json.Marshal(a.merge) - if _, err := js.Context().Publish(broker.DeliveryMergedSubject, body); err != nil { + if _, err := js.Context().Publish(broker.DeliveryMergedSubjectOf(theDelivery), body); err != nil { t.Fatal(err) } becomes(t, 15*time.Second, "the merge being kept", func() bool { diff --git a/cmd/mesh-controller/own_path_by_claim_test.go b/cmd/mesh-controller/own_path_by_claim_test.go index 86eb3230..a54bfd44 100644 --- a/cmd/mesh-controller/own_path_by_claim_test.go +++ b/cmd/mesh-controller/own_path_by_claim_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "strings" "testing" @@ -84,3 +85,9 @@ func TestBothClaimantsAreOnTheOwnPathAndGetTheRule(t *testing.T) { } } } + +// theDeliveryClaims is a catalogue whose one module claiming the seat is today's mesh-delivery, for the tests that +// announce a merge from it. +func theDeliveryClaims(context.Context) ([]inventory.Entry, error) { + return []inventory.Entry{claimingTheSeat("mesh-delivery")}, nil +} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 0b5fc231..f80a7d9d 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -209,7 +209,7 @@ func serve(ctx context.Context) (err error) { } // And the merges mesh-delivery made, kept a week with the ask each was made on: what registers a module that // moved between repositories on the operator's Approve (novox/hq ADR 0420 §7, issue 1025). - if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, + if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: open.inventory.Catalogued, kept: following{open: open}.resumeFirstBuilds}); err != nil { return err } diff --git a/internal/broker/streams.go b/internal/broker/streams.go index d297d917..5077fb20 100644 --- a/internal/broker/streams.go +++ b/internal/broker/streams.go @@ -448,8 +448,18 @@ var ControllerFollows = []string{ // the heads that Approve covered, kept a week to read the warrant a moved module's registration needs. var ControllerHears = []string{DeliveryMergedSubject} -// DeliveryMergedSubject is where mesh-delivery says it merged a pull request. -var DeliveryMergedSubject = moduleEventSubject("mesh-delivery", "merged") +// DeliveryMergedEvent is the event a module says when it merged a pull request. +const DeliveryMergedEvent = "merged" + +// DeliveryMergedSubject is where the delivery's owner says it merged a pull request: one wildcard subject, the +// module that claims the delivery seat being whichever the catalogue says (novox/hq ADR 0437 §3). Who said it is +// read from the subject, and whether it claims the seat is the controller's to check. +var DeliveryMergedSubject = moduleEventSubject("*", DeliveryMergedEvent) + +// DeliveryMergedSubjectOf is where one module says it merged a pull request. +func DeliveryMergedSubjectOf(module string) string { + return moduleEventSubject(module, DeliveryMergedEvent) +} // AsksSeat is the seat an ask is made on and its warrant heard from (novox/hq ADR 0259): the router's. const AsksSeat = "operator-channel" diff --git a/internal/link/delivery_merged_test.go b/internal/link/delivery_merged_test.go new file mode 100644 index 00000000..b24916c7 --- /dev/null +++ b/internal/link/delivery_merged_test.go @@ -0,0 +1,102 @@ +package link + +import ( + "bytes" + "context" + "log" + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" +) + +// The controller hears a merge's announcement from whichever module claims the delivery seat, so the subject is a +// wildcard and the emitter is read from it (novox/hq ADR 0437 §3, issue 1211). + +type keptMerges struct { + emitters []string + refuse bool +} + +func (k *keptMerges) KeepDeliveryMerge(_ context.Context, emitter string, _ []byte) error { + if k.refuse { + return ErrNotTheDeliverysWord + } + k.emitters = append(k.emitters, emitter) + return nil +} + +func mergeSays(t *testing.T, in *fakeInbound, to *settled, subject string) Control { + t.Helper() + m := in.sends(t, to, KindDeliveryMerged, map[string]any{"owner": "novox"}).(*fakeControl) + m.subject = subject + return m +} + +// 1211/AC1: a merged announcement on the subject of the module claiming the seat is read as the delivery's word and +// kept, as today's module's is (3: unchanged). +func TestAMergedAnnouncementIsKeptWhicheverModuleSaidIt(t *testing.T) { + s, in := serving() + kept := &keptMerges{} + if err := s.KeepsDeliveryMerges(kept); err != nil { + t.Fatal(err) + } + for _, module := range []string{"mesh-delivery-gate", "mesh-delivery"} { // 1211/AC1, 1211/AC3 + subject := "mesh.mod." + module + ".event.merged" + if kind, ok := kindOfSubject(subject); !ok || kind != KindDeliveryMerged { + t.Errorf("%s decodes to %q %v", subject, kind, ok) + } + if got, ok := DeliveryMergedEmitter(subject); !ok || got != module { + t.Errorf("%s: emitter %q %v", subject, got, ok) + } + to := &settled{} + s.act(t.Context(), mergeSays(t, in, to, subject)) + if !to.acked { + t.Errorf("%s: not taken: %+v", module, to) + } + } + if len(kept.emitters) != 2 || kept.emitters[0] != "mesh-delivery-gate" || kept.emitters[1] != "mesh-delivery" { + t.Errorf("the keeper was told the emitters %v", kept.emitters) + } + for _, subject := range []string{"mesh.mod.a.b.event.merged", "mesh.mod.x.event.merged.more", "mesh.seat.x.event.merged", "mesh.mod..event.merged"} { + if _, ok := DeliveryMergedEmitter(subject); ok { + t.Errorf("%s read as a merge announcement", subject) + } + } +} + +// 1211/AC2: an announcement from a module that claims no delivery seat is not the delivery's word: nothing is kept +// and the log names the module that said it. +func TestAMergedAnnouncementFromAModuleNotClaimingTheSeatIsNamedAndNotKept(t *testing.T) { + s, in := serving() + var said bytes.Buffer + s.log = log.New(&said, "", 0) + if err := s.KeepsDeliveryMerges(&keptMerges{refuse: true}); err != nil { + t.Fatal(err) + } + to := &settled{} + s.act(t.Context(), mergeSays(t, in, to, "mesh.mod.gitea.event.merged")) + if !to.rejected || to.acked || len(in.held) != 0 { + t.Errorf("settled %+v, held %d", to, len(in.held)) + } + if !strings.Contains(said.String(), "gitea") { + t.Errorf("the log does not name the module that said it: %q", said.String()) + } +} + +// 1211/AC4: the controller's consumer filter hears merged from any module, on one wildcard subject. +func TestTheControllerHearsMergedFromAnyModuleOnOneSubject(t *testing.T) { + if broker.DeliveryMergedSubject != "mesh.mod.*.event.merged" { + t.Errorf("the subject is %q", broker.DeliveryMergedSubject) + } + found := false + for _, s := range broker.ControllerHears { + found = found || s == broker.DeliveryMergedSubject + } + if !found { + t.Errorf("the controller hears %v", broker.ControllerHears) + } + if got := broker.DeliveryMergedSubjectOf("mesh-delivery-gate"); got != "mesh.mod.mesh-delivery-gate.event.merged" { + t.Errorf("a module's own subject is %q", got) + } +} diff --git a/internal/link/receive_nats.go b/internal/link/receive_nats.go index 31d6cc00..1ddeec42 100644 --- a/internal/link/receive_nats.go +++ b/internal/link/receive_nats.go @@ -296,8 +296,6 @@ func kindOfSubject(subject string) (string, bool) { return KindPullUpdated, true case broker.DecidedSubject: return KindDecided, true - case broker.DeliveryMergedSubject: - return KindDeliveryMerged, true case BuildOutcome(), BuildOutcomeOf(TheBuildMachineBefore): // A build's outcome is the role's event now, so it arrives on the events stream rather than // the control branch — and is acted on by the same handler, because what the controller does @@ -305,12 +303,30 @@ func kindOfSubject(subject string) (string, bool) { // runs (ADR 0190): the old builder still answers on the retired seat until it is unassigned. return KindBuilt, true } + if _, ok := DeliveryMergedEmitter(subject); ok { + return KindDeliveryMerged, true + } if _, ok := ProvisionerEmitter(subject); ok { return KindProvisioner, true } return "", false } +// DeliveryMergedEmitter is the module a merge's announcement came from, read from its subject +// (`mesh.mod..event.merged`); false for any other subject. The controller's own follow pattern, with `*` +// for the module, decodes too. +func DeliveryMergedEmitter(subject string) (string, bool) { + rest, ok := strings.CutPrefix(subject, "mesh.mod.") + if !ok { + return "", false + } + module, event, ok := strings.Cut(rest, ".event.") + if !ok || module == "" || strings.Contains(module, ".") || event != broker.DeliveryMergedEvent { + return "", false + } + return module, true +} + // ProvisionerEmitter is the module a provider's standing event came from, read from its subject // (`mesh.mod..event.provisioner.`); false for any other subject. The // controller's own follow pattern, with `*` for the module, decodes too. diff --git a/internal/link/serve.go b/internal/link/serve.go index 3eaabcfe..76f635f1 100644 --- a/internal/link/serve.go +++ b/internal/link/serve.go @@ -159,10 +159,16 @@ func (s *Server) Decides(d Decider) error { // DeliveryMergeKeeper is what the controller does with mesh-delivery's word that it merged a pull request (novox/hq // ADR 0420 §7): keep it, to be read when a moved module's build is taken in. +// +// The emitter is the module the announcement's subject names; a keeper that finds it does not claim the delivery seat +// answers ErrNotTheDeliverysWord (novox/hq ADR 0437 §3). type DeliveryMergeKeeper interface { - KeepDeliveryMerge(ctx context.Context, body []byte) error + KeepDeliveryMerge(ctx context.Context, emitter string, body []byte) error } +// ErrNotTheDeliverysWord is a keeper's answer to a merge's announcement from a module that claims no delivery seat. +var ErrNotTheDeliverysWord = errors.New("the module claims no delivery seat") + // KeepsDeliveryMerges says what to do about mesh-delivery's merges, and asks for them to be delivered. func (s *Server) KeepsDeliveryMerges(k DeliveryMergeKeeper) error { if err := s.inbound.Also(KindDeliveryMerged); err != nil { @@ -730,7 +736,15 @@ func (s *Server) deliveryMerged(ctx context.Context, m Control) { _ = m.Took() return } - err := s.deliveryMerges.KeepDeliveryMerge(ctx, m.Body()) + emitter, _ := DeliveryMergedEmitter(m.Subject()) + err := s.deliveryMerges.KeepDeliveryMerge(ctx, emitter, m.Body()) + if errors.Is(err, ErrNotTheDeliverysWord) { + // A merged event of a module that claims no delivery seat: not the delivery's word, so nothing is kept (and + // handing it back would not change that). + s.log.Printf("a word on a merge from %q is not the delivery's, and nothing is kept: %v", emitter, err) + _ = m.Drop() + return + } switch s.decide(ctx, m, "mesh-delivery's word on a merge", "", "", err) { case Hold: return