From ce72b67cef7dbdd1e894c829d4fac5d3b77c6566 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 11:20:42 +0100 Subject: [PATCH] mesh-grants: hear the condition events through the controller's durable consumer and acknowledge them; the provided mesh-wireguard manifest always wins (issue 1945) --- modules/mesh-grants/cmd/mesh-grants/main.go | 17 ++++++++----- modules/mesh-grants/grants_test.go | 7 ++++++ modules/mesh-grants/internal/grants/inputs.go | 25 +++++++++++-------- 3 files changed, 33 insertions(+), 16 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 093e67c..e232745 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -125,14 +125,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil }) } + // The controller makes this module's durable consumer on EVENTS from the manifest's consumes (filtered on the two + // condition subjects); it is pulled and every message acknowledged, so its pending never grows. Where it is not there + // yet the pass says so and hears again at the next change. f.Hear = func(handle func(string, []byte)) error { - for _, event := range grants.Consumes { - event := event - if _, err := nc.Subscribe(grants.EventSubject(event), func(m *nats.Msg) { handle(event, m.Data) }); err != nil { - return fmt.Errorf("hearing %s: %w", event, err) - } + cons, err := js.Consumer(ctx, "EVENTS", account.Node+"_"+account.Module) + if err != nil { + return fmt.Errorf("the consumer %s_%s on EVENTS: %w", account.Node, account.Module, err) } - return nil + _, 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) })) + return err } logger.Printf("started on the account %s: publishing memberships and creating the declared buckets, logs and streams the bus lacks; the user list, grants and any change to a held stream in shadow", account.User) f.Run(ctx) diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 5496f93..377a2ed 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -529,6 +529,13 @@ func TestGrants(t *testing.T) { // no gap, and its manifest is the one mesh-sdk provides (issue 1945). b.assign("b", "mesh-wireguard") in := grants.ReadInputs(b, &grants.Manifests{}) + // An indexer row of the same name never wins: the provided manifest is used and the row is said as a gap. + bw := fixture() + bw.manifestOf("mesh-wireguard", `{"module":"mesh-wireguard","version":"9","own-secrets":{"broker":"/x"}}`) + bw.assign("b", "mesh-wireguard") + if iw := grants.ReadInputs(bw, &grants.Manifests{}); !strings.HasPrefix(iw.Digests["mesh-wireguard"], "provided:") || !gapNamed(iw.Gaps, "manifest of mesh-wireguard") || len(iw.Refused) != 0 { + t.Errorf("an indexer manifest of mesh-wireguard won or went unsaid: %v %v", iw.Digests["mesh-wireguard"], iw.Gaps) + } if _, ok := in.Manifests["mesh-wireguard"]; !ok || gapNamed(in.Gaps, "manifest of mesh-wireguard") || !strings.HasPrefix(in.Digests["mesh-wireguard"], "provided:") { t.Errorf("mesh-wireguard's manifest is not taken from mesh-sdk: gaps %v", in.Gaps) } diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 49b747a..1d3a685 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -291,25 +291,30 @@ func ReadInputs(r Reader, cache *Manifests) Inputs { in.Manifests[module] = m in.Digests[module] = v.Manifest } - // A module the mesh provides itself has no repository the indexer reads: its manifest comes with mesh-sdk, as the - // controller's composer takes it (issue 1945, the private network on every machine). - for _, module := range modules { - if _, ok := in.Manifests[module]; ok || module != overlay.Name { - continue - } + // A module the mesh provides itself has no repository the indexer reads: its manifest comes with mesh-sdk, and the + // controller's Provide overwrites any row of it at every start, so the provided one always wins here too (issue 1945, + // the private network on every machine). An indexer manifest of the same name is said as a gap, never used. + if slices.Contains(modules, overlay.Name) { raw, err := json.Marshal(overlay.Manifest()) var m catalogue.Manifest if err == nil { m, err = catalogue.ParseManifestForInputs(raw) } + in.Gaps = slices.DeleteFunc(in.Gaps, func(g Gap) bool { return g.Input == "manifest of "+overlay.Name }) + in.Refused = slices.DeleteFunc(in.Refused, func(g Gap) bool { return g.Input == "manifest of "+overlay.Name }) + if _, held := in.Digests[overlay.Name]; held { + in.Gaps = append(in.Gaps, Gap{Input: "manifest of " + overlay.Name, Why: "the indexer also holds one, which is not used: the mesh provides this module's manifest itself, and the controller overwrites any row of it"}) + } if err != nil { - g := Gap{Input: "manifest of " + module, Why: "the manifest mesh-sdk provides is refused by its own reader: " + err.Error()} + g := Gap{Input: "manifest of " + overlay.Name, Why: "the manifest mesh-sdk provides is refused by its own reader: " + err.Error()} in.Refused = append(in.Refused, g) in.Gaps = append(in.Gaps, g) - continue + delete(in.Manifests, overlay.Name) + delete(in.Digests, overlay.Name) + } else { + in.Manifests[overlay.Name] = m + in.Digests[overlay.Name] = fmt.Sprintf("provided:sha256:%x", sha256.Sum256(raw)) } - in.Manifests[module] = m - in.Digests[module] = fmt.Sprintf("provided:sha256:%x", sha256.Sum256(raw)) } for _, module := range modules { if _, ok := in.Manifests[module]; !ok && !gapNamed(in.Gaps, "manifest of "+module) {