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

@@ -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)

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)
 		}

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) {