mesh-grants: create the durable consumers an account needs from the lease write, compared in shadow until then, so a module assigned after the controller leaves still hears what it consumes (issue 774, ADR 0561 §6)

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

@@ -18,6 +18,26 @@ type creating struct {
 	js  jetstream.JetStream
 }
 
+// CreateConsumer creates one durable consumer an account needs, by its stream and name (ADR 0561 §6): the create alone;
+// one the bus holds answers ErrHeld and is left as it is.
+func (c creating) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error {
+	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
+	defer cancel()
+	if _, err := c.js.Consumer(ctx, stream, cfg.Durable); err == nil {
+		return grants.ErrHeld
+	}
+	_, err := c.js.CreateConsumer(ctx, stream, cfg)
+	switch {
+	case err == nil:
+		return nil
+	case errors.Is(err, jetstream.ErrConsumerExists):
+		return grants.ErrHeld
+	case errors.Is(err, context.DeadlineExceeded):
+		return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err)
+	}
+	return err
+}
+
 func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error {
 	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
 	defer cancel()

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

@@ -112,7 +112,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	bus := newBusReader(ctx, js)
 	f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn},
-		Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js},
+		Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js}, Consumers: creating{ctx: ctx, js: js},
 		Objects: sealedObjects{ctx: ctx, js: js}, Logf: logger.Printf,
 		Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}}
 	f.Watch = bus