Merge pull request 'An older merge does not move a source' (#120) from fix/an-older-merge-does-not-move-a-source into main

The forge announces what it finds merged, and an old merge surfacing late moved the recorded head
backwards and rebuilt everything built from that repository, once per old merge. A merge made
before the source was last seen is history; one that says nothing about when is taken as news.
The catalogue now carries when each source was last seen. The bus-records test follows #116:
a module's tools are every one under its own name.

cmd/mesh-controller/order_test.go

@@ -3,6 +3,7 @@ package main
 import (
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/novox/mesh-controller/internal/catalogue"
 	"github.com/novox/mesh-controller/internal/inventory"
@@ -90,3 +91,18 @@ func TestAMergeMatchesTheSourcesBuiltFromIt(t *testing.T) {
 		}
 	}
 }
+
+// A merge made before the source was last seen is history: it does not move the source, and a
+// merge that says nothing about when it was made is taken as news.
+func TestAMergeOlderThanTheLastLookIsHistory(t *testing.T) {
+	seen := time.Date(2026, 9, 28, 3, 0, 0, 0, time.UTC)
+	if !isHistory("2026-09-28T02:00:00Z", seen) {
+		t.Fatal("an older merge was taken as news")
+	}
+	if isHistory("2026-09-28T04:00:00Z", seen) {
+		t.Fatal("a newer merge was taken as history")
+	}
+	if isHistory("", seen) || isHistory("2026-09-28T02:00:00Z", time.Time{}) {
+		t.Fatal("a merge or a source with no time on it was refused")
+	}
+}

cmd/mesh-controller/push.go

@@ -753,8 +753,31 @@ func raiseTheBus(ctx context.Context, inv *inventory.Inventory, address string)
 	if err := broker.RaiseSeats(js, inventory.MeshSeats(), holders); err != nil {
 		return err
 	}
-	fmt.Printf("the bus at %s has its streams, and %d machine(s) can hear a declaration\n",
+	// And how every module hears what it consumes. Derived from the same records the user list is
-		broker.BareAddress(address), len(names))
+	// composed from, so a module the mesh grants a consumer's subjects has that consumer waiting.
+	// Done on every raise, not only when a credential is issued: every module moved onto this bus
+	// by the rollout was issued on the old one, and came up with nothing to bind to (2026-09-28).
+	records, err := inv.BusRecords(ctx)
+	if err != nil {
+		return err
+	}
+	users, err := broker.Users(records)
+	if err != nil {
+		return err
+	}
+	hearing := 0
+	for _, p := range users {
+		consumer, needed := broker.ConsumerFor(p)
+		if !needed {
+			continue
+		}
+		if err := js.EnsureConsumer(consumer); err != nil {
+			return fmt.Errorf("how %s on %s hears what it consumes: %w", p.Module, p.Node, err)
+		}
+		hearing++
+	}
+	fmt.Printf("the bus at %s has its streams, %d machine(s) can hear a declaration, and %d module(s) "+
+		"can hear what they consume\n", broker.BareAddress(address), len(names), hearing)
 	return nil
 }
 

cmd/mesh-controller/upgrades.go

@@ -240,6 +240,13 @@ func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error {
 		if e.Source.BuiltFrom == m.Commit {
 			continue
 		}
+		// **A merge older than the last look at the source is history, not a move.** The forge
+		// announces what it finds merged, and an old merge surfacing late would otherwise move the
+		// recorded head backwards and rebuild everything built from that repository, once per old
+		// merge (2026-09-28).
+		if isHistory(m.MergedAt, e.Source.Seen) {
+			continue
+		}
 		if err := inv.SourceMoved(ctx, e.Manifest.Module, m.Commit); err != nil {
 			return notNow(err)
 		}
@@ -362,3 +369,16 @@ func standsOnModule(e inventory.Entry, module string, against map[string][]strin
 	}
 	return false
 }
+
+// isHistory is whether a merge made at mergedAt predates the last time the source was seen. A merge
+// with no time on it is taken as news: refusing it would silence a forge that says less.
+func isHistory(mergedAt string, seen time.Time) bool {
+	if mergedAt == "" || seen.IsZero() {
+		return false
+	}
+	at, err := time.Parse(time.RFC3339, mergedAt)
+	if err != nil {
+		return false
+	}
+	return at.Before(seen)
+}

internal/broker/nats.go

@@ -297,11 +297,18 @@ func PermissionsFor(p Principal) (Permissions, error) {
 			}
 		}
 
+		// 2c. Its own consumer, which it **pulls**: the runtime asks for the next message and is
+		//     answered on its own inbox, so what it needs is to ask about the consumer and to ask it
+		//     for messages — its own consumer's name, and no other's. Pulled rather than pushed
+		//     because that is the one shape a runtime's client binds without creating anything; the
+		//     controller and the hosts are pushed to. Named here rather than through ConsumerFor,
+		//     which asks for these permissions to build the consumer and would ask forever. A
+		//     subject for a consumer that turns out not to exist grants nothing anybody can use.
+		pub = append(pub,
+			"$JS.API.CONSUMER.INFO."+consumerStream(p)+"."+consumerDurable(p),
+			"$JS.API.CONSUMER.MSG.NEXT."+consumerStream(p)+"."+consumerDurable(p))
+
 		// 3. Seats it holds: full participation.
-		// Its consumer's name, not ConsumerFor: that asks for these permissions to build the
-		// consumer, and would ask forever. A subject for a consumer that turns out not to exist
-		// grants nothing anybody can use.
-		sub = append(sub, "_DELIVER."+consumerDurable(p))
 		for _, s := range p.Holds {
 			// Taking work from the role's queue: the worker consumer it binds (asked about,
 			// delivered on, acknowledged), each on the seat's own stream. The first machine to

internal/broker/nats_test.go

@@ -338,3 +338,24 @@ func admits(pattern, subject []string) bool {
 	}
 	return len(pattern) == len(subject)
 }
+
+// A module pulls its own consumer — asks about it, asks it for messages — and no other module's.
+func TestAModulePullsItsOwnConsumerAndNoOthers(t *testing.T) {
+	p, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "audit",
+		Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
+	for _, want := range []string{"$JS.API.CONSUMER.INFO.EVENTS.one_audit", "$JS.API.CONSUMER.MSG.NEXT.EVENTS.one_audit"} {
+		if !slices.Contains(p.Publish, want) {
+			t.Errorf("a module cannot bind its own consumer: %v lacks %s", p.Publish, want)
+		}
+	}
+	for _, s := range p.Publish {
+		if strings.Contains(s, "CONSUMER.") && !strings.HasSuffix(s, ".one_audit") {
+			t.Errorf("a module may reach another consumer: %s", s)
+		}
+	}
+	for _, s := range p.Subscribe {
+		if strings.HasPrefix(s, "_DELIVER.") {
+			t.Errorf("a module is granted a push delivery it never binds: %s", s)
+		}
+	}
+}

internal/broker/testdata/composed.conf

@@ -37,18 +37,18 @@ accounts {
         subscribe: { allow: ["_DELIVER.one", "_INBOX.node.one.>", "mesh.node.one.declare"] }
       } }
       { user: "one.telegram", password: "$2a$11$tttttttttttttttttttttt", permissions: {
-        publish: { allow: ["$JS.ACK.EVENTS.one_telegram.>", "$JS.ACK.SEAT_TELEGRAM_SENDER.SEAT_TELEGRAM_SENDER_worker.>", "$JS.API.CONSUMER.INFO.SEAT_TELEGRAM_SENDER.SEAT_TELEGRAM_SENDER_worker", "mesh.seat.telegram-sender.event.delivered", "mesh.seat.telegram-sender.event.failed"] }
+        publish: { allow: ["$JS.ACK.EVENTS.one_telegram.>", "$JS.ACK.SEAT_TELEGRAM_SENDER.SEAT_TELEGRAM_SENDER_worker.>", "$JS.API.CONSUMER.INFO.EVENTS.one_telegram", "$JS.API.CONSUMER.INFO.SEAT_TELEGRAM_SENDER.SEAT_TELEGRAM_SENDER_worker", "$JS.API.CONSUMER.MSG.NEXT.EVENTS.one_telegram", "mesh.seat.telegram-sender.event.delivered", "mesh.seat.telegram-sender.event.failed"] }
-        subscribe: { allow: ["_DELIVER.SEAT_TELEGRAM_SENDER_worker", "_DELIVER.one_telegram", "_INBOX.one.telegram.>", "mesh.mod.telegram.tool.>", "mesh.seat.telegram-sender.accept.send"] }
+        subscribe: { allow: ["_DELIVER.SEAT_TELEGRAM_SENDER_worker", "_INBOX.one.telegram.>", "mesh.mod.telegram.tool.>", "mesh.seat.telegram-sender.accept.send"] }
         allow_responses: { max: 1, ttl: "1m" }
       } }
       { user: "two.audit", password: "$2a$11$aaaaaaaaaaaaaaaaaaaaaa", permissions: {
-        publish: { allow: ["$JS.ACK.EVENTS.two_audit.>"] }
+        publish: { allow: ["$JS.ACK.EVENTS.two_audit.>", "$JS.API.CONSUMER.INFO.EVENTS.two_audit", "$JS.API.CONSUMER.MSG.NEXT.EVENTS.two_audit"] }
-        subscribe: { allow: ["_DELIVER.two_audit", "_INBOX.two.audit.>", "mesh.mod.audit.tool.>", "mesh.mod.shop.event.order.placed"] }
+        subscribe: { allow: ["_INBOX.two.audit.>", "mesh.mod.audit.tool.>", "mesh.mod.shop.event.order.placed"] }
         allow_responses: { max: 1, ttl: "1m" }
       } }
       { user: "two.shop", password: "$2a$11$ssssssssssssssssssssss", permissions: {
-        publish: { allow: ["$JS.ACK.EVENTS.two_shop.>", "mesh.mod.shop.event.order.placed", "mesh.seat.telegram-sender.accept.send"] }
+        publish: { allow: ["$JS.ACK.EVENTS.two_shop.>", "$JS.API.CONSUMER.INFO.EVENTS.two_shop", "$JS.API.CONSUMER.MSG.NEXT.EVENTS.two_shop", "mesh.mod.shop.event.order.placed", "mesh.seat.telegram-sender.accept.send"] }
-        subscribe: { allow: ["_DELIVER.two_shop", "_INBOX.two.shop.>", "mesh.mod.shop.tool.>"] }
+        subscribe: { allow: ["_INBOX.two.shop.>", "mesh.mod.shop.tool.>"] }
         allow_responses: { max: 1, ttl: "1m" }
       } }
     ]

internal/builder/mirror.go

@@ -180,6 +180,20 @@ func (r Registry) MirrorImage(ctx context.Context, from, repository string) (str
 	if err != nil {
 		return "", err
 	}
+	// **Already held is already mirrored.** A base is named by digest, and a digest this registry
+	// holds under the module's repository is the same bytes whatever upstream would say — so
+	// upstream is not asked. Asked every build, the public hub's anonymous pull limit was reached
+	// on the first merge that rebuilt a whole catalogue (2026-09-28), and every module whose base
+	// lives there failed on a copy it did not need.
+	if strings.HasPrefix(where.reference, "sha256:") {
+		held, err := r.has(ctx, "http://"+r.Address+"/v2/"+repository+"/manifests/"+where.reference)
+		if err != nil {
+			return "", fmt.Errorf("asking %s whether it holds %s: %w", r.Address, from, err)
+		}
+		if held {
+			return r.Address + "/" + repository + "@" + where.reference, nil
+		}
+	}
 	src := &source{client: r.client()}
 	digest, err := r.copyManifest(ctx, src, where, where.reference, repository)
 	if err != nil {

internal/builder/mirror_test.go

@@ -103,6 +103,12 @@ func (m *theMeshsRegistry) handler() http.Handler {
 		m.mu.Lock()
 		defer m.mu.Unlock()
 		switch {
+		case r.Method == http.MethodHead && strings.Contains(r.URL.Path, "/manifests/"):
+			if _, ok := m.manifests[r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:]]; ok {
+				w.WriteHeader(http.StatusOK)
+			} else {
+				w.WriteHeader(http.StatusNotFound)
+			}
 		case r.Method == http.MethodHead && strings.Contains(r.URL.Path, "/blobs/"):
 			if _, ok := m.blobs[r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:]]; ok {
 				w.WriteHeader(http.StatusOK)
@@ -219,3 +225,27 @@ func TestATagBeforeTheDigestIsNotPartOfTheRepository(t *testing.T) {
 		t.Fatalf("got %+v", got)
 	}
 }
+
+// A base this registry already holds by digest is not asked of upstream at all: the public hub
+// limits anonymous pulls, and a catalogue rebuilt on one merge asked it once per module.
+func TestABaseAlreadyHeldIsNotAskedOfUpstream(t *testing.T) {
+	src, indexDigest, _ := anUpstreamRegistry(t)
+	dst := &theMeshsRegistry{blobs: map[string][]byte{}, manifests: map[string][]byte{}}
+	dstServer := httptest.NewServer(dst.handler())
+	defer dstServer.Close()
+	address := strings.TrimPrefix(dstServer.URL, "http://")
+	r := Registry{Address: address, HTTP: src.Client()}
+	host := strings.TrimPrefix(src.URL, "http://")
+	if _, err := r.MirrorImage(context.Background(), host+"/library/thing:latest", "hello-web/server"); err != nil {
+		t.Fatal(err)
+	}
+	// Upstream gone: the pinned base is answered from what the mesh holds.
+	src.Close()
+	reference, err := r.MirrorImage(context.Background(), host+"/library/thing@"+indexDigest, "hello-web/server")
+	if err != nil {
+		t.Fatalf("a base the registry holds was asked of an upstream that is gone: %v", err)
+	}
+	if reference != address+"/hello-web/server@"+indexDigest {
+		t.Fatalf("pinned as %q", reference)
+	}
+}

internal/inventory/busrecords_test.go

@@ -86,9 +86,11 @@ func TestAnAssignedModuleBecomesAUserWithWhatItDeclared(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
+		// Its tools are every one under its own name — the list in the manifest is a person's
+		// vocabulary for asking, not the module's permission to answer.
 		if !granted(perms.Publish, "mesh.mod.shop.event.order.placed") ||
 			!granted(perms.Publish, "mesh.seat.telegram-sender.accept.send") ||
-			!granted(perms.Subscribe, "mesh.mod.shop.tool.price") {
+			!granted(perms.Subscribe, "mesh.mod.shop.tool.>") {
 			t.Fatalf("one.shop's authority is not what it declared: %+v", perms)
 		}
 	}

internal/inventory/catalogue.go

@@ -8,6 +8,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/novox/mesh-controller/internal/catalogue"
@@ -38,6 +39,9 @@ type Source struct {
 	BuiltFrom string
 	// Head is the newest commit the source is known to have.
 	Head string
+	// Seen is when the source was last looked at — by a build, by hand, or by the forge saying it
+	// moved. What a late report of an older move is judged against.
+	Seen time.Time
 }
 
 // Current reports whether what the mesh holds is what the source last had.
@@ -936,12 +940,13 @@ func (i *Inventory) Catalogued(ctx context.Context) ([]Entry, error) {
 		`select m.name, m.manifest,
 		        coalesce(m.source, ''), m.source_path, m.source_seat, coalesce(m.ref, ''),
 		        coalesce(m.built_from, ''), coalesce(m.source_head, ''),
+		        coalesce(m.source_seen, to_timestamp(0)),
 		        coalesce(array_agg(n.name order by n.name) filter (where n.name is not null), '{}')
 		 from module m
 		 left join assignment a on a.module = m.name
 		 left join node n on n.id = a.node
 		 group by m.name, m.manifest, m.source, m.source_path, m.source_seat, m.ref, m.built_from,
-		          m.source_head
+		          m.source_head, m.source_seen
 		 order by m.name`)
 	if err != nil {
 		return nil, err
@@ -955,9 +960,12 @@ func (i *Inventory) Catalogued(ctx context.Context) ([]Entry, error) {
 		var source Source
 		var on []string
 		if err := rows.Scan(&name, &raw, &source.Repository, &source.Path, &source.Seat, &source.Ref,
-			&source.BuiltFrom, &source.Head, &on); err != nil {
+			&source.BuiltFrom, &source.Head, &source.Seen, &on); err != nil {
 			return nil, err
 		}
+		if source.Seen.Unix() == 0 {
+			source.Seen = time.Time{}
+		}
 		var m catalogue.Manifest
 		if err := json.Unmarshal(raw, &m); err != nil {
 			return nil, err

internal/link/events.go

@@ -128,6 +128,8 @@ type SourceMoved struct {
 	Commit   string `json:"merge_commit_sha"`
 	CloneURL string `json:"clone_url"`
 	HTMLURL  string `json:"html_url"`
+	// MergedAt is when the forge merged it, RFC 3339. What decides whether this is news.
+	MergedAt string `json:"merged_at"`
 }
 
 type Upgraded struct {