diff --git a/internal/broker/buildbacklog.go b/internal/broker/buildbacklog.go new file mode 100644 index 00000000..2c99b283 --- /dev/null +++ b/internal/broker/buildbacklog.go @@ -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-.accept.` 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 +} diff --git a/internal/broker/buildbacklog_test.go b/internal/broker/buildbacklog_test.go new file mode 100644 index 00000000..f39261a4 --- /dev/null +++ b/internal/broker/buildbacklog_test.go @@ -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) + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index b9b0de1c..0d7d3706 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -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 diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index 5d4ce16e..43a916c7 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -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: {