Grant the watchman the build seats' accept subjects and the view the seat workers' information, each by name, so the backlog and the waiting-ask pulse read live (issue 1887 A8)

2026-10-23 16:32:32 +02:00
parent 24ad859629
commit bb0d5e17db
4 changed files with 143 additions and 1 deletions
@@ -0,0 +1,46 @@
+package broker
+
+import "git.novox.be/novox/mesh-sdk/go/delivery"
+
+// The two modules that read the build seats' queues, each by name (novox/hq ADR 0525 §2, issue 1887 A8): the watchman hears
+// the asks put on the seats (a plain subscription, no consumer made) to mirror each waiting ask as a pulse, and the view
+// reads the shared workers' counts and the oldest waiting ask to show the backlog. Neither may publish an ask: only the
+// builder and the checker ask, by their seat use. Nothing here is a wildcard over a stream or a worker.
+const (
+	// ViewModule is the module of the delivery view (mesh-delivery-view).
+	ViewModule = "mesh-delivery-view"
+)
+
+// BuildSeatAsksHeard are the subjects the watchman subscribes to: `mesh.seat.node-build-<language>.accept.<kind>` of every
+// build seat and kind (delivery.AskSubject), each by name.
+func BuildSeatAsksHeard() []string {
+	var out []string
+	for _, lang := range delivery.AskLanguages {
+		for _, kind := range delivery.AskKinds {
+			if s, err := delivery.AskSubject(lang, kind); err == nil {
+				out = append(out, s)
+			}
+		}
+	}
+	return out
+}
+
+// BuildBacklogReads are the API subjects the view requests: the information of each build seat's stream and of each of its
+// shared workers (delivery.AskStream, delivery.AskWorker), and the message of the stream by sequence. Reads only: no
+// message is pulled, none acknowledged, none published.
+func BuildBacklogReads() []string {
+	var out []string
+	for _, lang := range delivery.AskLanguages {
+		stream, err := delivery.AskStream(lang)
+		if err != nil {
+			continue
+		}
+		out = append(out, "$JS.API.STREAM.INFO."+stream, "$JS.API.STREAM.MSG.GET."+stream)
+		for _, kind := range delivery.AskKinds {
+			if w, err := delivery.AskWorker(lang, kind); err == nil {
+				out = append(out, "$JS.API.CONSUMER.INFO."+stream+"."+w)
+			}
+		}
+	}
+	return out
+}
@@ -0,0 +1,90 @@
+package broker
+
+import (
+	"strings"
+	"testing"
+)
+
+func hasSubject(list []string, s string) bool {
+	for _, x := range list {
+		if x == s {
+			return true
+		}
+	}
+	return false
+}
+
+func composedFor(t *testing.T, module string) (pub, sub []string) {
+	t.Helper()
+	perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: module})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return perms.Publish, perms.Subscribe
+}
+
+func TestTheWatchmanHearsTheBuildSeatsAsksByName(t *testing.T) { // 1887/A8b
+	_, sub := composedFor(t, WatchmanModule)
+	for _, lang := range []string{"go", "node", "image"} {
+		for _, kind := range []string{"delivery", "check", "recheck"} {
+			if s := "mesh.seat.node-build-" + lang + ".accept." + kind; !hasSubject(sub, s) {
+				t.Errorf("the watchman cannot subscribe %s", s)
+			}
+		}
+	}
+	for _, s := range sub {
+		if strings.Contains(s, "node-build") && (strings.HasSuffix(s, ">") || strings.HasSuffix(s, "*")) {
+			t.Errorf("a wildcard over the build seats: %s", s)
+		}
+	}
+	pub, _ := composedFor(t, WatchmanModule)
+	for _, s := range pub {
+		if strings.Contains(s, "node-build") || strings.Contains(s, "SEAT_NODE_BUILD") {
+			t.Errorf("the watchman may publish on %s", s)
+		}
+	}
+	perms, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: WatchmanModule})
+	if MayPublish(perms, "mesh.seat.node-build-go.accept.check") {
+		t.Error("the watchman may publish an ask")
+	}
+	// no other module hears them
+	if _, other := composedFor(t, "shop"); containsAny(other, "node-build") {
+		t.Errorf("another module hears the build seats: %v", other)
+	}
+}
+
+func containsAny(list []string, sub string) bool {
+	for _, x := range list {
+		if strings.Contains(x, sub) {
+			return true
+		}
+	}
+	return false
+}
+
+func TestTheViewReadsTheBuildSeatsWorkersByNameAndPublishesNoAsk(t *testing.T) { // 1887/A8a
+	pub, sub := composedFor(t, ViewModule)
+	for _, s := range []string{
+		"$JS.API.STREAM.INFO.SEAT_NODE_BUILD_GO", "$JS.API.STREAM.MSG.GET.SEAT_NODE_BUILD_GO",
+		"$JS.API.CONSUMER.INFO.SEAT_NODE_BUILD_GO_CHECK_worker", "$JS.API.CONSUMER.INFO.SEAT_NODE_BUILD_NODE_RECHECK_worker",
+		"$JS.API.CONSUMER.INFO.SEAT_NODE_BUILD_IMAGE_DELIVERY_worker", "$JS.API.STREAM.MSG.GET.SEAT_NODE_BUILD_IMAGE",
+	} {
+		if !hasSubject(pub, s) {
+			t.Errorf("the view cannot request %s", s)
+		}
+	}
+	for _, s := range append(append([]string{}, pub...), sub...) {
+		if strings.Contains(s, "mesh.seat.node-build") || (strings.Contains(s, "NODE_BUILD") && (strings.Contains(s, "MSG.NEXT") || strings.Contains(s, "$JS.ACK") || strings.Contains(s, "CONSUMER.CREATE"))) {
+			t.Errorf("the view may publish an ask or pull and acknowledge: %s", s)
+		}
+	}
+	perms, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: ViewModule})
+	for _, s := range []string{"mesh.seat.node-build-go.accept.check", "$JS.API.CONSUMER.MSG.NEXT.SEAT_NODE_BUILD_GO.SEAT_NODE_BUILD_GO_CHECK_worker"} {
+		if MayPublish(perms, s) {
+			t.Errorf("the view may publish %s", s)
+		}
+	}
+	if other, _ := composedFor(t, "shop"); containsAny(other, "SEAT_NODE_BUILD") {
+		t.Errorf("another module reads the build seats: %v", other)
+	}
+}
@@ -855,6 +855,12 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		if p.Module == WatchmanModule {
 			sub = append(sub, BusConnectSubject, BusDisconnectSubject)
 			pub = append(pub, BusConnzSubject)
+			// And the asks put on the build seats, to mirror each waiting one as a pulse (issue 1887 A8b).
+			sub = append(sub, BuildSeatAsksHeard()...)
+		}
+		// The view alone reads the build seats' workers for the backlog of asks (issue 1887 A8a).
+		if p.Module == ViewModule {
+			pub = append(pub, BuildBacklogReads()...)
 		}
 		// 1. Its own namespace: it publishes its events there and serves its tools there. Nothing
 		//    else may publish into it, so an event's source is a fact the server enforces rather
@@ -44,7 +44,7 @@ accounts {
      } }
      { user: "one.mesh-delivery-watchman", password: "$2a$11$wwwwwwwwwwwwwwwwwwwwww", permissions: {
        publish: { allow: ["$JS.ACK.EVENTS.one_mesh-delivery-watchman.>", "$JS.API.CONSUMER.INFO.EVENTS.one_mesh-delivery-watchman", "$JS.API.CONSUMER.MSG.NEXT.EVENTS.one_mesh-delivery-watchman", "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.one.mesh-delivery-watchman", "mesh.bus.connz"] }
-        subscribe: { allow: ["$SRV.INFO", "$SRV.INFO.mesh-delivery-watchman", "$SRV.INFO.mesh-delivery-watchman.>", "$SRV.PING", "$SRV.PING.mesh-delivery-watchman", "$SRV.PING.mesh-delivery-watchman.>", "$SRV.STATS", "$SRV.STATS.mesh-delivery-watchman", "$SRV.STATS.mesh-delivery-watchman.>", "_INBOX.one.mesh-delivery-watchman.>", "mesh.assignment.one.mesh-delivery-watchman", "mesh.bus.connect", "mesh.bus.disconnect", "mesh.mod.mesh-delivery-watchman.call.>", "mesh.mod.mesh-delivery-watchman.tool.>"] }
+        subscribe: { allow: ["$SRV.INFO", "$SRV.INFO.mesh-delivery-watchman", "$SRV.INFO.mesh-delivery-watchman.>", "$SRV.PING", "$SRV.PING.mesh-delivery-watchman", "$SRV.PING.mesh-delivery-watchman.>", "$SRV.STATS", "$SRV.STATS.mesh-delivery-watchman", "$SRV.STATS.mesh-delivery-watchman.>", "_INBOX.one.mesh-delivery-watchman.>", "mesh.assignment.one.mesh-delivery-watchman", "mesh.bus.connect", "mesh.bus.disconnect", "mesh.mod.mesh-delivery-watchman.call.>", "mesh.mod.mesh-delivery-watchman.tool.>", "mesh.seat.node-build-go.accept.check", "mesh.seat.node-build-go.accept.delivery", "mesh.seat.node-build-go.accept.recheck", "mesh.seat.node-build-image.accept.check", "mesh.seat.node-build-image.accept.delivery", "mesh.seat.node-build-image.accept.recheck", "mesh.seat.node-build-node.accept.check", "mesh.seat.node-build-node.accept.delivery", "mesh.seat.node-build-node.accept.recheck"] }
        allow_responses: { max: 1, ttl: "1m" }
      } }
      { user: "one.nats", password: "$2a$11$bbbbbbbbbbbbbbbbbbbbbb", permissions: {