Hear a merge's announcement from whichever module claims the delivery seat (issue 1211)

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 {

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

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

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 {

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 {

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

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
 	}

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"

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

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.<module>.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.<module>.event.provisioner.<failing|recovered|retirement>`); false for any other subject. The
 // controller's own follow pattern, with `*` for the module, decodes too.

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