Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3756bb3460 | ||
|
|
60be9c5360 | ||
|
|
da31bcb11e | ||
|
|
f03e7b33c9 | ||
|
|
0baf727f36 | ||
|
|
94dd49a968 | ||
|
|
83a298e7e0 | ||
|
|
220b79f5cd |
@@ -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",
|
||||
broker.BareAddress(address), len(names))
|
||||
// And how every module hears what it consumes. Derived from the same records the user list is
|
||||
// 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
|
||||
}
|
||||
|
||||
|
||||
+27
-10
@@ -181,6 +181,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
for _, seat := range meshSeatsTheControllerUses {
|
||||
pub = append(pub, "mesh.seat."+seat+".accept.>")
|
||||
}
|
||||
// Every module's tools: **the control plane is the way in** (novox/hq ADR 0095). A person
|
||||
// or an agent asks through it and every question passes one process where an audit
|
||||
// belongs — so it, alone among principals, may call any tool by name. The first `ask` on
|
||||
// the new bus was refused the publish (2026-09-28).
|
||||
pub = append(pub, "mesh.mod.*.tool.>")
|
||||
|
||||
// The two events it reacts to, and its ack subject on the stream they arrive from
|
||||
// (streams.go). **Each named, not a pattern**: `mesh.mod.*.event.>` would make the
|
||||
@@ -266,9 +271,13 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
for _, e := range p.Emits {
|
||||
pub = append(pub, own+".event."+e)
|
||||
}
|
||||
for _, t := range p.Serves {
|
||||
sub = append(sub, own+".tool."+t)
|
||||
}
|
||||
// Every tool under its own name, not a list: the tools a module serves are what its code
|
||||
// answers, and a second copy of that list in the manifest would be a second source of
|
||||
// truth for the mesh to keep in step (2026-09-28: every module that served a tool was
|
||||
// refused the subscription, because none had written the list twice). Nothing is given
|
||||
// away — no other principal may subscribe this namespace, and a caller's authority is
|
||||
// still granted per tool, by name, on the publish side.
|
||||
sub = append(sub, own+".tool.>")
|
||||
|
||||
// 2. What it consumes, by the emitter's own subject — an event is addressed to its
|
||||
// emitter, because the emitter's identity is the meaning (ADR 0118).
|
||||
@@ -288,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
|
||||
@@ -348,9 +364,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
return Permissions{
|
||||
Publish: pub,
|
||||
Subscribe: sub,
|
||||
// Only something that serves is ever answering. A pure consumer is granted nothing here.
|
||||
AllowResponses: p.Kind == KindModule && (len(p.Serves) > 0 || len(p.Holds) > 0) ||
|
||||
p.Kind == KindController,
|
||||
// A module answers what it was asked — a tool call reaches it on its own namespace, so the
|
||||
// authority is bounded by having been asked — and so does the controller. A node and a
|
||||
// person are never asked anything, and are granted nothing here.
|
||||
AllowResponses: p.Kind == KindModule || p.Kind == KindController,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package broker
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
@@ -89,17 +90,30 @@ func TestAnInboxIsScopedToItsOwner(t *testing.T) {
|
||||
}
|
||||
|
||||
// A responder answers on the caller's inbox, which it has no permission for. allow_responses is
|
||||
// what makes a scoped inbox workable at all — the authority is bounded by having been asked.
|
||||
func TestOnlySomethingThatServesMayAnswer(t *testing.T) {
|
||||
serving, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "billing",
|
||||
Serves: []string{"status"}, PasswordHash: "x"})
|
||||
if !serving.AllowResponses {
|
||||
t.Fatal("a module serving a tool cannot answer the caller's inbox")
|
||||
}
|
||||
consumer, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "audit",
|
||||
// what makes a scoped inbox workable at all — the authority is bounded by having been asked. A
|
||||
// module is asked on its own namespace and may answer; a node and a person are never asked.
|
||||
func TestOnlyWhatCanBeAskedMayAnswer(t *testing.T) {
|
||||
module, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "audit",
|
||||
Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
|
||||
if consumer.AllowResponses {
|
||||
t.Fatal("a pure consumer was granted the right to answer, which nothing asked it to do")
|
||||
if !module.AllowResponses {
|
||||
t.Fatal("a module cannot answer a tool call on its own namespace")
|
||||
}
|
||||
node, _ := PermissionsFor(Principal{Kind: KindNode, Node: "one", PasswordHash: "x"})
|
||||
if node.AllowResponses {
|
||||
t.Fatal("a node was granted the right to answer, and nothing asks a node anything")
|
||||
}
|
||||
}
|
||||
|
||||
// A module serves every tool under its own name, and no other module's.
|
||||
func TestAModuleServesItsOwnNamespaceAndNoOthers(t *testing.T) {
|
||||
p, _ := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "gitea", PasswordHash: "x"})
|
||||
if !slices.Contains(p.Subscribe, "mesh.mod.gitea.tool.>") {
|
||||
t.Fatalf("a module may not serve its own tools: %v", p.Subscribe)
|
||||
}
|
||||
for _, s := range p.Subscribe {
|
||||
if strings.HasPrefix(s, "mesh.mod.") && !strings.HasPrefix(s, "mesh.mod.gitea.") {
|
||||
t.Fatalf("a module may subscribe another's namespace: %s", s)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -324,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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+9
-7
@@ -24,7 +24,7 @@ accounts {
|
||||
jetstream: enabled
|
||||
users = [
|
||||
{ user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
|
||||
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "_INBOX.enrol.>", "mesh.control.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>"] }
|
||||
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "_INBOX.enrol.>", "mesh.control.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>"] }
|
||||
subscribe: { allow: ["$JS.API.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.gitea.event.pull.merged", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
@@ -37,17 +37,19 @@ 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"] }
|
||||
subscribe: { allow: ["_DELIVER.SEAT_TELEGRAM_SENDER_worker", "_DELIVER.one_telegram", "_INBOX.one.telegram.>", "mesh.mod.telegram.tool.status", "mesh.seat.telegram-sender.accept.send"] }
|
||||
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", "_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.>"] }
|
||||
subscribe: { allow: ["_DELIVER.two_audit", "_INBOX.two.audit.>", "mesh.mod.shop.event.order.placed"] }
|
||||
publish: { allow: ["$JS.ACK.EVENTS.two_audit.>", "$JS.API.CONSUMER.INFO.EVENTS.two_audit", "$JS.API.CONSUMER.MSG.NEXT.EVENTS.two_audit"] }
|
||||
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"] }
|
||||
subscribe: { allow: ["_DELIVER.two_shop", "_INBOX.two.shop.>"] }
|
||||
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: ["_INBOX.two.shop.>", "mesh.mod.shop.tool.>"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
]
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user