From 745c30e06865a88044fc2906fb5e485b1d1479c7 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:03:02 +0100 Subject: [PATCH] DLV-G1, DLV-W1: read the bus module's events by the mesh-broker contract's field names, on its own stream or the seat's subject MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The seat's definition (mesh-contracts meshbroker) names local-name and storage-type and puts the events on mesh.seat.mesh-broker.event.; design 25 § 12.3 keeps them on the bus module's own stream. Both are read until that is ruled. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-delivery-watchman/consumes.go | 11 +++++- .../cmd/mesh-delivery-watchman/main_test.go | 2 +- .../internal/watch/busstate.go | 9 ++++- modules/mesh-grants/grants_test.go | 3 +- .../mesh-grants/internal/grants/bindings.go | 38 ++++++++++++++----- 5 files changed, 47 insertions(+), 16 deletions(-) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go index f6788e91..dcdd34bf 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -8,6 +8,7 @@ import ( "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/membership" "git.novox.be/novox/mesh-sdk/go/process" @@ -39,13 +40,13 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, return } w.Accepted(a) - case strings.HasPrefix(subject, "mesh.stream.") && strings.HasSuffix(subject, ".event."+eventCreated): + case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated): if s, err := watch.ReadBrokerSaid(body); err != nil { logf("%v; left", err) } else { w.BrokerCreated(s) } - case strings.HasPrefix(subject, "mesh.stream.") && strings.HasSuffix(subject, ".event."+eventRefused): + case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventRefused): if s, err := watch.ReadBrokerSaid(body); err != nil { logf("%v; left", err) } else { @@ -108,3 +109,9 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco } }() } + +// brokerSubject is a subject the bus module's created or refused arrives on: its own stream's, or the seat mesh-broker's +// (where the seat's definition puts it; only its holder publishes there). +func brokerSubject(subject string) bool { + return strings.HasPrefix(subject, "mesh.stream.") || strings.HasPrefix(subject, "mesh.seat."+catalogue.BrokerSeat+".event.") +} diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index 7ce9b82c..654c757e 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -61,7 +61,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 t.Fatalf("an accepted manifest no longer asking for l left %v", out.findings) } consumed(w, "mesh.stream.nats.event.refused", []byte(`{"module":"m","local":"l"}`), say) - consumed(w, "mesh.stream.nats.event.created", []byte(`{"module":"m","local":"l"}`), say) + consumed(w, "mesh.seat.mesh-broker.event.created", []byte(`{"module":"m","local-name":"l"}`), say) if len(out.findings) != 0 { t.Fatalf("created left %v", out.findings) } diff --git a/modules/mesh-delivery-watchman/internal/watch/busstate.go b/modules/mesh-delivery-watchman/internal/watch/busstate.go index ff1cc5bf..0141ca8e 100644 --- a/modules/mesh-delivery-watchman/internal/watch/busstate.go +++ b/modules/mesh-delivery-watchman/internal/watch/busstate.go @@ -35,8 +35,10 @@ const ( // create was refused. It stands until the seat mesh-broker's definition is generated in mesh-contracts. type BrokerSaid struct { Module string `json:"module"` - Local string `json:"local"` - Why string `json:"why,omitempty"` + // Local is the entry: the contract's `local-name` (mesh-contracts meshbroker.Said), or `local` where a body names it so. + Local string `json:"local,omitempty"` + LocalName string `json:"local-name,omitempty"` + Why string `json:"why,omitempty"` } // busStateID is the signal's and the findings' id of one entry: `.`. @@ -145,6 +147,9 @@ func ReadBrokerSaid(body []byte) (BrokerSaid, error) { if err := json.Unmarshal(body, &s); err != nil { return BrokerSaid{}, fmt.Errorf("the bus module's word cannot be read: %w", err) } + if s.Local == "" { + s.Local = s.LocalName + } if s.Module == "" || s.Local == "" { return BrokerSaid{}, fmt.Errorf("the bus module's word names no module or no entry") } diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index c1e51ce6..fb2896e0 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -2261,7 +2261,8 @@ func TestGrants(t *testing.T) { if got := bindingsOn("b.mod-x"); len(got) != 1 || got[0].Bucket != "mod-x_records-2" { t.Errorf("after recreated: %+v", got) } - if !hear("mesh.stream.nats.event.dropped", membership.BusBinding{Local: "records"}) { + // The seat's own subject, and the contract's field names (mesh-contracts meshbroker.Said), are read alike. + if changed, err := p.HearBroker("mesh.seat.mesh-broker.event.dropped", []byte(`{"module":"mod-x","local-name":"records"}`), p.Brokers()); err != nil || !changed { t.Fatal("dropped changed nothing") } p.Compose() diff --git a/modules/mesh-grants/internal/grants/bindings.go b/modules/mesh-grants/internal/grants/bindings.go index 3a4f2542..badc7e04 100644 --- a/modules/mesh-grants/internal/grants/bindings.go +++ b/modules/mesh-grants/internal/grants/bindings.go @@ -6,6 +6,7 @@ import ( "sort" "strings" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/membership" "mesh-grants/internal/buscompose" ) @@ -39,11 +40,15 @@ func BoundKey(module string) string { return BoundPrefix + module } // It stands until the seat mesh-broker's definition is generated in mesh-contracts (ADR 0613 §1, §5; design 26): this // part then reads the generated type, with the same fields. type BrokerEvent struct { - V int `json:"v"` - Module string `json:"module"` - Local string `json:"local"` - Storage string `json:"storage"` - Template string `json:"template,omitempty"` + V int `json:"v,omitempty"` + Module string `json:"module"` + // LocalName and StorageType are the seat's contract's names (mesh-contracts meshbroker.Said); Local and Storage the + // same read where an older body names them so. + LocalName string `json:"local-name,omitempty"` + Local string `json:"local,omitempty"` + StorageType string `json:"storage-type,omitempty"` + Storage string `json:"storage,omitempty"` + Template string `json:"template,omitempty"` // Binding is the entry's names on the bus: whole on `created` and `recreated`; on `dropped` only its local name counts. Binding membership.BusBinding `json:"binding"` } @@ -61,6 +66,12 @@ type Bound struct { func brokerEvent(event string, brokers map[string]bool) (string, bool) { emitter, name := "", "" switch { + case strings.HasPrefix(event, "mesh.seat."+catalogue.BrokerSeat+".event."): + // The seat's own subject: only its holder publishes there, so the emitter is the holder. + name = strings.TrimPrefix(event, "mesh.seat."+catalogue.BrokerSeat+".event.") + for b := range brokers { + emitter = b + } case buscompose.OnAModuleStream(event): // mesh.stream..event. rest := strings.TrimPrefix(event, "mesh.stream.") @@ -93,10 +104,8 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b if err := json.Unmarshal(body, &e); err != nil { return false, fmt.Errorf("the bus module's %s does not read: %w", name, err) } - local := e.Local - if local == "" { - local = e.Binding.Local - } + local := firstOf(e.LocalName, e.Local, e.Binding.Local) + storage := firstOf(e.StorageType, e.Storage) if e.Module == "" || local == "" || strings.Contains(e.Module, ".") { return false, fmt.Errorf("the bus module's %s names no module or no entry (%q, %q)", name, e.Module, local) } @@ -115,7 +124,7 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b bind := e.Binding bind.Local = local if bind.Storage == "" { - bind.Storage = e.Storage + bind.Storage = storage } if bind.Template == "" { bind.Template = e.Template @@ -245,3 +254,12 @@ func bindingStreams(b membership.BusBinding) []string { } return out } + +func firstOf(values ...string) string { + for _, v := range values { + if v != "" { + return v + } + } + return "" +}