diff --git a/go/process/account_test.go b/go/process/account_test.go index 7c7002b..10c30b2 100644 --- a/go/process/account_test.go +++ b/go/process/account_test.go @@ -257,3 +257,39 @@ func (h *heldSubscriber) deliver() { h.mu.Unlock() handle(&nats.Msg{Subject: subject}) } + +// 1242: the module's own credential gives the names it reaches on the bus (its events' subjects, its own +// stream's, its durable consumer's) and the options it dials with. +func TestModuleAccountNamesAndDialOptions(t *testing.T) { + a := Account{URL: "nats://x:4222", Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"} + for _, c := range []struct{ name, got, want string }{ + {"event subject", a.EventSubject("put"), "mesh.mod.mesh-delivery-teller.event.put"}, + {"own stream subject", a.StreamSubject("forge.x.pull"), "mesh.stream.mesh-delivery-teller.event.forge.x.pull"}, + {"durable consumer", a.EventConsumerName(), "n1_mesh-delivery-teller"}, + } { + if c.got != c.want { + t.Errorf("%s: got %q, want %q", c.name, c.got, c.want) + } + } + if _, err := ReadBrokerAccount(func(string) string { return " " }); err == nil || !strings.Contains(err.Error(), BrokerFileEnv) { + t.Errorf("no broker file set: %v", err) + } + var o nats.Options + for _, opt := range a.DialOptions() { + if err := opt(&o); err != nil { + t.Fatal(err) + } + } + if o.Name != "n1.mesh-delivery-teller" || o.User != "n1.mesh-delivery-teller" || + o.MaxReconnect != -1 || !o.RetryOnFailedConnect || o.Secure { + t.Errorf("dial options: %+v", o) + } + a.Fingerprint = "sha256:ab" + o = nats.Options{} + for _, opt := range a.DialOptions() { + _ = opt(&o) + } + if !o.Secure || o.TLSConfig == nil { + t.Errorf("a fingerprint did not pin the connection") + } +} diff --git a/go/process/module.go b/go/process/module.go new file mode 100644 index 0000000..8310bfe --- /dev/null +++ b/go/process/module.go @@ -0,0 +1,91 @@ +package process + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// What a module's own principal reaches on the bus beyond a process account's queue (novox/hq issue 1242): its +// own credential, its own stream and its durable consumer on the shared events stream. These helpers are the one +// place a long-running part of a module (a daemon the module's delivery runs) learns them, so no part hand-writes +// the connection, the subject spaces or the consumer's name. + +// EventsStream is the shared stream every module's events (mesh.mod..event.>) are on; a module's consumer +// on it is the durable _ the controller makes (mesh-controller internal/broker/nats.go +// consumerStream, consumerDurable). +const EventsStream = "EVENTS" + +// BrokerFileEnv names the environment variable holding the path of the module's own broker credential. +const BrokerFileEnv = "MESH_BROKER_FILE" + +// ReadBrokerAccount reads the module's own credential from the file BrokerFileEnv names in getenv. +func ReadBrokerAccount(getenv func(string) string) (Account, error) { + path := strings.TrimSpace(getenv(BrokerFileEnv)) + if path == "" { + return Account{}, errors.New(BrokerFileEnv + " is not set, so there is no credential to hear the bus with") + } + return ReadAccount(path) +} + +// DialOptions is the option list Dial connects with: named ., as the account's user on its own +// inbox prefix, reconnecting without limit and retrying a first connect that fails, pinned to the account's +// fingerprint when it carries one; opts follow, so a caller may add a handler or override one. +func (a Account) DialOptions(opts ...nats.Option) []nats.Option { + all := []nats.Option{ + nats.Name(a.Node + "." + a.Module), + nats.UserInfo(a.User, a.Password), + nats.CustomInboxPrefix(a.InboxPrefix()), + nats.MaxReconnects(-1), + nats.RetryOnFailedConnect(true), + } + if strings.TrimSpace(a.Fingerprint) != "" { + all = append(all, nats.Secure(PinnedToFingerprint(a.Fingerprint))) + } + return append(all, opts...) +} + +// Dial connects as the module's own principal and opens JetStream on the connection. The caller closes the +// connection. +func (a Account) Dial(opts ...nats.Option) (*nats.Conn, jetstream.JetStream, error) { + nc, err := nats.Connect(a.URL, a.DialOptions(opts...)...) + if err != nil { + return nil, nil, fmt.Errorf("connecting to the bus: %w", err) + } + js, err := jetstream.New(nc) + if err != nil { + nc.Close() + return nil, nil, err + } + return nc, js, nil +} + +// EventSubject is where the module's event lands on the shared events stream: mesh.mod..event.. +func (a Account) EventSubject(event string) string { return "mesh.mod." + a.Module + ".event." + event } + +// StreamSubject is where the module's event lands on a stream of its own: mesh.stream..event. +// (mesh-controller internal/broker/modstreams.go ModuleStreamSubject). +func (a Account) StreamSubject(event string) string { + return "mesh.stream." + a.Module + ".event." + event +} + +// EventConsumerName is the module's durable consumer on a stream it is granted: _. +func (a Account) EventConsumerName() string { return a.Node + "_" + a.Module } + +// EventConsumer binds the module's durable consumer on stream (EventsStream, or a stream of another module it is +// granted). +func (a Account) EventConsumer(ctx context.Context, js jetstream.JetStream, stream string) (jetstream.Consumer, error) { + return js.Consumer(ctx, stream, a.EventConsumerName()) +} + +// HearEvents pulls from cons, handing each message to handle, and says every error the bus reports while +// hearing through logf ("hearing the bus: ..."). The caller stops the returned context. +func HearEvents(cons jetstream.Consumer, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) { + return cons.Consume(handle, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { + logf("hearing the bus: %v", err) + })) +}