DLV-G1: the grants part writes the bus module's binding into the module's memberships on created, rewrites it on recreated, removes it on dropped, kept in memberships (ADR 0595 §1, §7)

The part hears the bus module's own stream on its durable consumer there. That consumer did
not exist: since the controller ended (ADR 0590) nobody created a module's consumer on a
module stream it hears (AccountConsumers left them to the controller), so it is created now
with every account consumer, and a pass that created consumers wakes one more pass so the
part's own hearing opens at once. Bindings are kept per module under bound.<module>.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

modules/mesh-grants/cmd/mesh-grants/integration_test.go

@@ -96,9 +96,17 @@ func TestTheGrantsPartWhole(t *testing.T) {
 		verdict, _ := delivery.EncodeManifestVerdict(delivery.ManifestVerdict{Module: module, State: delivery.VerdictAccepted, Judge: "j", At: time.Now()})
 		put(grants.ReadVerdicts, digest, json.RawMessage(verdict))
 	}
-	manifest("nats", `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"}}`)
+	manifest("nats", `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"},`+
+		`"emits":["created","done","dropped","recreated"],"streams":[{"name":"broker-events","events":["created","dropped","recreated"],"max-age-days":30,"max-mib":256}]}`)
 	manifest("mod-c", `{"module":"mod-c","version":"1","own-secrets":{"broker":"/var/lib/c/broker"},"consumes":["nats.done"]}`)
-	pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "a", Modules: []delivery.PlacedModule{{Module: "nats"}, {Module: "mod-c"}}})
+	// The part itself, placed as on the mesh, so its own membership names the bus module's stream it hears (DLV-G1).
+	own, err := os.ReadFile("../../module.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	manifest(grants.Module, string(own))
+	pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "a", Modules: []delivery.PlacedModule{{Module: "nats"}, {Module: "mod-c"},
+		{Module: grants.Module}}})
 	if _, err := kvs[grants.ReadPlacement].Put(ctx, "placement.a", pl); err != nil {
 		t.Fatal(err)
 	}
@@ -189,6 +197,41 @@ func TestTheGrantsPartWhole(t *testing.T) {
 	if n := said.Load(); n != 0 {
 		t.Errorf("%d memberships were said on their subject", n)
 	}
+	// DLV-G1, 1966/AC34 (novox/hq ADR 0595 §1, §7): the bus module's created, said on its own stream, reaches the part on
+	// its durable consumer there, and the binding it carries is in the module's membership; dropped takes it out.
+	until("the part's durable consumer on the bus module's stream", func() bool {
+		e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a."+grants.Module)
+		if err != nil {
+			return false
+		}
+		m, err := membership.DecodeGrantsMembership(e.Value())
+		if err != nil || len(m.Hears) == 0 {
+			return false
+		}
+		_, err = js.Consumer(ctx, m.Hears[0].Stream, m.Hears[0].Consumer)
+		return err == nil
+	})
+	bindingOf := func() []membership.BusBinding {
+		e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c")
+		if err != nil {
+			return nil
+		}
+		m, _ := membership.DecodeGrantsMembership(e.Value())
+		return m.Bindings
+	}
+	sayBroker := func(event string, bind membership.BusBinding) {
+		body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-c", Local: "kept", Storage: "key-value", Template: "state", Binding: bind})
+		if _, err := js.Publish(ctx, "mesh.stream.nats.event."+event, body); err != nil {
+			t.Fatal(err)
+		}
+	}
+	sayBroker("created", membership.BusBinding{Local: "kept", Storage: "key-value", Template: "state", Bucket: "mod-c_kept"})
+	until("mod-c's binding in its membership after created", func() bool {
+		b := bindingOf()
+		return len(b) == 1 && b[0].Bucket == "mod-c_kept"
+	})
+	sayBroker("dropped", membership.BusBinding{Local: "kept"})
+	until("mod-c's binding gone after dropped", func() bool { return len(bindingOf()) == 0 })
 	for _, bucket := range []string{"mesh-controller_lease", grants.Module + "_lease", grants.Module + "_takeover"} {
 		if _, err := js.KeyValue(ctx, bucket); !errors.Is(err, jetstream.ErrBucketNotFound) {
 			t.Errorf("the bucket %s: %v, want none", bucket, err)

modules/mesh-grants/cmd/mesh-grants/main.go

@@ -121,16 +121,42 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// The part makes this module's own durable consumer on EVENTS from the manifest's consumes (filtered on the two
 	// condition subjects and the vault's account-issued; ADR 0561 §6), as it makes every module's; 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. The vault's bus-accounts is watched beside it, so a hash is composed either way.
+	//
+	// Beside EVENTS, the module streams its consumes reach (the bus module's own stream, whose created, recreated and
+	// 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{}
 	f.Hear = func(handle func(string, []byte)) error {
-		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)
+		consume := func(stream, name string) error {
+			if opened[stream] {
+				return nil
+			}
+			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) {
+				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
 		}
-		_, 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
+		if err := consume("EVENTS", account.Node+"_"+account.Module); err != nil {
+			return err
+		}
+		hears, err := grants.OwnHears(ownStore{conn}, account.Node)
+		if err != nil {
+			return err
+		}
+		for _, h := range hears {
+			if err := consume(h.Stream, h.Consumer); err != nil {
+				return err
+			}
+		}
+		return nil
 	}
 	logger.Printf("started on the account %s: publishing memberships, writing grants and the user list sealed to the bus node, which it follows, and creating the account consumers and the declared buckets, logs and streams the bus lacks; a change to a held stream is only said", account.User)
 	f.Run(ctx)