From 25a704a1f562090f619e33a20347fea8be847350 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:32:27 +0100 Subject: [PATCH] Grants part: its one pull of its durable on the shared events stream is the module's event hub: the owners' views, the seat mesh-catalog's verdicts and the other seats' events are handlers of it (B7c finding 1; issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- modules/mesh-grants/cmd/mesh-grants/main.go | 62 +++++++++++-------- modules/mesh-grants/grants_test.go | 10 +++ .../mesh-grants/internal/grants/rootfree.go | 19 +++++- 3 files changed, 63 insertions(+), 28 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index deb2ca28..012ac6cb 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -142,41 +142,40 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // dropped carry the bindings: ADR 0595 §1, §7), each on the durable consumer its own membership names; one not // there yet is opened at the next change, and one opened is never opened twice. opened := map[string]bool{} + // The module has one durable consumer of EVENTS: one hub pulls it and hands each message to the part that hears it + // (the owners' views events, the seat mesh-catalog's verdicts, the other seats' events the follower hears), started + // once all are registered. + hub := process.NewEventHub(logger.Printf) + defer hub.Stop() + registered := false f.Hear = func(handle func(string, []byte)) error { - consume := func(stream, name string) error { - if opened[stream] { - return nil + if !registered { + registered = true + // An owner's views event goes to the follower of its state, which acknowledges it (firstviews.go). + if err := hub.Handle(views.EventsPattern, hearViews); err != nil { + return err } - cons, err := js.Consumer(ctx, stream, name) - if err != nil { - return fmt.Errorf("the consumer %s on %s: %w", name, stream, err) - } - if _, err := cons.Consume(func(m jetstream.Msg) { - // An owner's views event goes to the follower of its state, which acknowledges it (firstviews.go). - if strings.Contains(m.Subject(), ".event."+views.EventPrefix) { - hearViews(m) - return - } - // The seat mesh-catalog's verdict events go to the verdicts' first view (ADR 0622 §2), which takes each at its revision. - if event, ok := strings.CutPrefix(m.Subject(), "mesh.seat."+meshcatalog.Seat+".event."); ok && - (event == meshcatalog.EventManifestAccepted || event == meshcatalog.EventManifestRefused) { - if err := fv.HearVerdict(ctx, event, m.Data(), logger.Printf); err != nil { + // The seat mesh-catalog's verdict events go to the verdicts' first view (ADR 0622 §2), which takes each at its revision. + for _, event := range []string{meshcatalog.EventManifestAccepted, meshcatalog.EventManifestRefused} { + if err := hub.Handle("mesh.seat."+meshcatalog.Seat+".event."+event, func(m jetstream.Msg) { + if err := fv.HearVerdict(ctx, strings.TrimPrefix(m.Subject(), "mesh.seat."+meshcatalog.Seat+".event."), m.Data(), logger.Printf); err != nil { logger.Printf("views: %s is not a verdict event, dropped: %v", m.Subject(), err) _ = m.Term() return } _ = m.Ack() - return + }); err != nil { + return err + } + } + // The other seats' events the part follows its passes by (Consumes). + for _, subject := range grants.SeatEventSubjects() { + if err := hub.Handle(subject, func(m jetstream.Msg) { handle(m.Subject(), m.Data()); _ = m.Ack() }); err != nil { + return err } - handle(m.Subject(), m.Data()) - _ = m.Ack() - }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })); err != nil { - return err } - opened[stream] = true - return nil } - if err := consume("EVENTS", account.Node+"_"+account.Module); err != nil { + if err := hub.Start(ctx, js, account); err != nil { return err } hears, err := grants.OwnHears(ownStore{conn}, account.Node) @@ -184,9 +183,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } for _, h := range hears { - if err := consume(h.Stream, h.Consumer); err != nil { + if opened[h.Stream] { + continue + } + cons, err := js.Consumer(ctx, h.Stream, h.Consumer) + if err != nil { + return fmt.Errorf("the consumer %s on %s: %w", h.Consumer, h.Stream, err) + } + if _, err := cons.Consume(func(m jetstream.Msg) { + handle(m.Subject(), m.Data()) + _ = m.Ack() + }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })); err != nil { return err } + opened[h.Stream] = true } return nil } diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index cd5fbfc5..69ea7ee3 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -2214,6 +2214,16 @@ func TestGrants(t *testing.T) { t.Errorf("the part consumes %s", c) } } + // The hub's seat handlers are the seats' events of Consumes on the seats' own subjects, the verdicts' and views' apart. + for _, subject := range grants.SeatEventSubjects() { + seat, event, _ := strings.Cut(strings.TrimPrefix(subject, "mesh.seat."), ".event.") + if !slices.Contains(grants.Consumes, seat+"."+event) || strings.Contains(subject, "mesh-catalog") { + t.Errorf("%s is no seat event of Consumes beside the verdicts and views", subject) + } + } + if n := len(grants.SeatEventSubjects()); n != 6 { + t.Errorf("the part hears %d seats' events, want the keeper's two, the vault's one and the broker's three", n) + } for _, c := range b.calls { if strings.Contains(c, "mesh-controller") { t.Errorf("the part called %s", c) diff --git a/modules/mesh-grants/internal/grants/rootfree.go b/modules/mesh-grants/internal/grants/rootfree.go index 01e7c6df..a270dcc0 100644 --- a/modules/mesh-grants/internal/grants/rootfree.go +++ b/modules/mesh-grants/internal/grants/rootfree.go @@ -42,8 +42,23 @@ const EventAccountIssued = "mesh-vault." + buscompose.AccountIssuedEvent // Consumes is every event the part hears, as its manifest's consumes name them. // The seat mesh-broker's created, recreated and dropped are heard as every seat's event, `.` on EVENTS (the // lead's ruling, registry milestone 13:0x), never on the bus module's own stream. -var Consumes = append([]string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, meshbroker.Seat + "." + BrokerCreated, - meshbroker.Seat + "." + BrokerDropped, meshbroker.Seat + "." + BrokerRecreated}, viewsEvents()...) +var Consumes = append(slices.Clone(seatEvents), viewsEvents()...) + +// seatEvents are the events of seats the part follows its passes by, as `.`. +var seatEvents = []string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, meshbroker.Seat + "." + BrokerCreated, + meshbroker.Seat + "." + BrokerDropped, meshbroker.Seat + "." + BrokerRecreated} + +// SeatEventSubjects are the subjects of the seats' events the part follows its passes by besides the verdicts' and the +// views': the keeper's condition events, the vault's account-issued and the seat mesh-broker's, each `.` of +// Consumes on the seat's own subject. +func SeatEventSubjects() []string { + var out []string + for _, c := range seatEvents { + seat, event, _ := strings.Cut(c, ".") + out = append(out, "mesh.seat."+seat+".event."+event) + } + return out +} // viewsEvents are the owners' events the part follows its first views by: `.views.`, but the verdicts by // the indexer's manifest-accepted and manifest-refused (VerdictEvents, ADR 0622 §2).