go/membership/membership.go

@@ -0,0 +1,165 @@
+// Package membership is what the mesh issues an assignment to serve and to reach (novox/hq ADR 0160), as a
+// process reads it from the bus: the Membership shape, the one subject it is published on, the stream that
+// holds it, and the bus address helpers a reader needs. The controller composes it and the route proxy reads
+// it; both use this one definition, so there is no second copy to drift (novox/hq issue 1633).
+package membership
+
+import (
+	"encoding/json"
+	"strings"
+)
+
+// AssignmentsStream holds every assignment's membership, the newest per subject.
+const AssignmentsStream = "ASSIGNMENTS"
+
+// Subject is the one address a runtime derives for itself: where its own membership is published, from
+// the two names its credential carries. Everything else is in the membership.
+func Subject(node, module string) string {
+	return "mesh.assignment." + node + "." + module
+}
+
+// DirectGetSubject is the subject-addressed direct get of a membership from its stream: the one request an
+// account may make of the stream.
+func DirectGetSubject(node, module string) string {
+	return "$JS.API.DIRECT.GET." + AssignmentsStream + "." + Subject(node, module)
+}
+
+// Membership is one assignment's subjects: what this instance of a module on this machine serves,
+// and what it may reach.
+type Membership struct {
+	Node   string `json:"node"`
+	Module string `json:"module"`
+	// Serves is every address a tool of this instance answers on. `{tool}` stands for the tool's
+	// own name. An address with a queue is shared with the module's other instances.
+	Serves []Served `json:"serves"`
+	// Seats is every verb of a seat this instance holds, at the subject the seat's callers use.
+	Seats []SeatServed `json:"seats,omitempty"`
+	// Emits is where an event of this module lands; `{event}` stands for the event's name.
+	Emits string `json:"emits"`
+	// Reaches is each tool this module may call, `<module>.<tool>`, to the subjects that reach it.
+	Reaches map[string][]string `json:"reaches,omitempty"`
+	// Tools is where this instance answers what it serves.
+	Tools string `json:"tools"`
+	// Receives is what this assignment is given for each requirement it receives, by requirement (novox/hq
+	// ADR 0167): kept as JSON because the catalogue owns the shape of a contribution.
+	Receives map[string]json.RawMessage `json:"receives,omitempty"`
+	// Mesh is every machine's address on the private network (novox/hq ADR 0167).
+	Mesh []string `json:"mesh,omitempty"`
+	// State is every bucket this module's code may reach, and whether it may write it (novox/hq ADR 0201).
+	State []StateIssued `json:"state,omitempty"`
+	// Logs is every log this module's code may reach (novox/hq ADR 0297).
+	Logs []LogIssued `json:"logs,omitempty"`
+	// SeatTraffic is what this module's code may submit, say, hear, take, ask, answer and read on seats
+	// that name their caller or their kind (novox/hq ADR 0259 §3).
+	SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"`
+	// Streams are this module's own streams, Hears the module streams it hears (novox/hq ADR 0363).
+	Streams []StreamIssued `json:"streams,omitempty"`
+	Hears   []HeardIssued  `json:"hears,omitempty"`
+	// Waits are the event families a session may wait on through the mesh MCP server (novox/hq ADR 0367 §2).
+	Waits []WaitIssued `json:"waits,omitempty"`
+}
+
+// Served is one address a tool is answered on.
+type Served struct {
+	Subject string `json:"subject"`
+	Queue   string `json:"queue,omitempty"`
+}
+
+// SeatServed is one verb of a held seat, where its callers ask.
+type SeatServed struct {
+	Seat    string `json:"seat"`
+	Verb    string `json:"verb"`
+	Subject string `json:"subject"`
+}
+
+// StateIssued is one bucket an assignment may reach, by the name its module uses for it (novox/hq ADR 0201).
+// Key, when set, is the one key it may reach (novox/hq ADR 0260): this machine's.
+type StateIssued struct {
+	Name   string `json:"name"`
+	Bucket string `json:"bucket"`
+	Writes bool   `json:"writes,omitempty"`
+	Key    string `json:"key,omitempty"`
+}
+
+// LogIssued is one log an assignment may reach, by the name its module uses for it (novox/hq ADR 0297).
+type LogIssued struct {
+	Name    string `json:"name"`
+	Stream  string `json:"stream"`
+	Subject string `json:"subject"`
+	Writes  bool   `json:"writes"`
+}
+
+// Worker is one durable consumer a holder pulls a seat's work from.
+type Worker struct {
+	Stream   string
+	Consumer string
+	Filter   string
+}
+
+// SeatTraffic is one module's seat traffic beyond tools. Publish and Subscribe are subject patterns, in the
+// server's wildcards.
+type SeatTraffic struct {
+	Publish   []string   `json:"publish,omitempty"`
+	Subscribe []string   `json:"subscribe,omitempty"`
+	Answers   []string   `json:"answers,omitempty"`
+	Workers   []Worker   `json:"workers,omitempty"`
+	Records   []string   `json:"records,omitempty"`
+	Kinds     []KindHeld `json:"kinds,omitempty"`
+}
+
+// KindHeld is one kind of a kinded bench and who holds it.
+type KindHeld struct {
+	Seat         string   `json:"seat"`
+	Kind         string   `json:"kind"`
+	Module       string   `json:"module"`
+	Node         string   `json:"node"`
+	Capabilities []string `json:"capabilities,omitempty"`
+}
+
+// StreamIssued is one of a module's own streams as its membership names it (ADR 0363 §2).
+type StreamIssued struct {
+	Stream  string   `json:"stream"`
+	Events  []string `json:"events"`
+	Subject string   `json:"subject"`
+}
+
+// HeardIssued is one module stream an assignment hears (ADR 0363 §8).
+type HeardIssued struct {
+	Stream   string `json:"stream"`
+	Consumer string `json:"consumer"`
+}
+
+// WaitIssued is one family as a membership issues it: the name a session asks for, and its subject.
+type WaitIssued struct {
+	Event   string `json:"event"`
+	Subject string `json:"subject"`
+}
+
+// CredentialIn reads the credential out of a bus address, `[scheme://]user:password@host:port`, splitting on
+// the last at-sign so a password containing one stays whole, and returns the address without it.
+func CredentialIn(address string) (user, password, bare string) {
+	at := strings.LastIndex(address, "@")
+	if at < 0 {
+		return "", "", address
+	}
+	scheme := ""
+	rest := address[:at]
+	if i := strings.Index(rest, "://"); i >= 0 {
+		scheme, rest = rest[:i+3], rest[i+3:]
+	}
+	user, password, _ = strings.Cut(rest, ":")
+	return user, password, scheme + address[at+1:]
+}
+
+// BareAddress is a bus address with any credential stripped, for something that only needs to know
+// whether a server is answering there.
+func BareAddress(address string) string {
+	_, _, bare := CredentialIn(address)
+	if bare == "" {
+		return address
+	}
+	if strings.Contains(bare, "://") {
+		return bare
+	}
+	return "nats://" + bare
+}

go/membership/membership_test.go

@@ -0,0 +1,76 @@
+package membership
+
+import (
+	"encoding/json"
+	"reflect"
+	"testing"
+)
+
+func TestACredentialIsReadOutOfABusAddress(t *testing.T) {
+	for _, c := range []struct{ in, user, password, bare string }{
+		{"nats://controller:secret@127.0.0.1:4222", "controller", "secret", "nats://127.0.0.1:4222"},
+		{"controller:secret@127.0.0.1:4222", "controller", "secret", "127.0.0.1:4222"},
+		{"nats://127.0.0.1:4222", "", "", "nats://127.0.0.1:4222"},
+		{"127.0.0.1:4222", "", "", "127.0.0.1:4222"},
+		// A password containing an at-sign: split on the last one.
+		{"nats://controller:a@b@127.0.0.1:4222", "controller", "a@b", "nats://127.0.0.1:4222"},
+	} {
+		user, password, bare := CredentialIn(c.in)
+		if user != c.user || password != c.password || bare != c.bare {
+			t.Errorf("%q read as %q/%q at %q; wanted %q/%q at %q", c.in, user, password, bare, c.user, c.password, c.bare)
+		}
+	}
+}
+
+func TestABareAddressCarriesNoCredentialAndAScheme(t *testing.T) {
+	for in, want := range map[string]string{
+		"nats://controller:secret@10.0.0.1:4222": "nats://10.0.0.1:4222",
+		"controller:secret@10.0.0.1:4222":        "nats://10.0.0.1:4222",
+		"nats://10.0.0.1:4222":                   "nats://10.0.0.1:4222",
+		"10.0.0.1:4222":                          "nats://10.0.0.1:4222",
+	} {
+		if got := BareAddress(in); got != want {
+			t.Errorf("BareAddress(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+func TestAMembershipIsPublishedAndReadAtTheSubjectDerivedFromTheTwoNames(t *testing.T) {
+	if got := Subject("ace", "route-proxy"); got != "mesh.assignment.ace.route-proxy" {
+		t.Errorf("subject: %q", got)
+	}
+	if got := DirectGetSubject("ace", "route-proxy"); got != "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.ace.route-proxy" {
+		t.Errorf("direct get: %q", got)
+	}
+	if AssignmentsStream != "ASSIGNMENTS" {
+		t.Errorf("stream: %q", AssignmentsStream)
+	}
+}
+
+// The shape on the wire is the controller's: a membership as issued reads back whole, with the routes kept raw.
+func TestAMembershipRoundTripsAsIssued(t *testing.T) {
+	in := Membership{Node: "ace", Module: "route-proxy", Emits: "e", Tools: "t",
+		Serves:      []Served{{Subject: "s", Queue: "q"}},
+		Receives:    map[string]json.RawMessage{"route": json.RawMessage(`[{"name":"a"}]`)},
+		Mesh:        []string{"10.10.0.7"},
+		State:       []StateIssued{{Name: "n", Bucket: "b", Writes: true, Key: "ace"}},
+		SeatTraffic: &SeatTraffic{Publish: []string{"p"}, Kinds: []KindHeld{{Seat: "channel", Kind: "tg"}}}}
+	raw, err := json.Marshal(in)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var out Membership
+	if err := json.Unmarshal(raw, &out); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(in, out) {
+		t.Fatalf("round trip changed it:\n%+v\n%+v", in, out)
+	}
+	var wire map[string]json.RawMessage
+	_ = json.Unmarshal(raw, &wire)
+	for _, k := range []string{"receives", "mesh", "seat-traffic"} {
+		if _, ok := wire[k]; !ok {
+			t.Errorf("field %q is not on the wire under that name", k)
+		}
+	}
+}

go/process/conn.go

@@ -40,7 +40,7 @@ var refusedPublish = regexp.MustCompile(`(?i)Publish to "(\S+)"`)
 // other certificate), authenticated as its user and password, on the inbox prefix _INBOX.<user> and no
 // other, reconnecting without limit as the tool runner does (novox/hq issue 1192: a process outlives a
 // bus restart, never giving up the way the client library's own default of 60 attempts would). Ported
-// from mesh-controller internal/broker/jetstream.go:106-132 (DialPinned, PinnedToFingerprint) since this
+// from mesh-controller internal/broker/jetstream.go (DialPinned, PinnedToFingerprint), which now uses this one since this
 // package may not import the controller.
 //
 // An ErrorHandler among opts is chained behind this package's own, which watches every async error for a
@@ -78,7 +78,7 @@ func (c *Conn) options(opts []nats.Option) []nats.Option {
 		nats.MaxReconnects(-1),
 	}
 	if c.account.Fingerprint != "" {
-		all = append(all, nats.Secure(pinnedToFingerprint(c.account.Fingerprint)))
+		all = append(all, nats.Secure(PinnedToFingerprint(c.account.Fingerprint)))
 	}
 	all = append(all, opts...)
 	return append(all, nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
@@ -140,9 +140,11 @@ func (c *Conn) publishRefused(err error) bool {
 	return len(cancels) > 0
 }
 
-// pinnedToFingerprint accepts exactly the certificate with this SHA-256 and no other — the leaf alone, its
+// PinnedToFingerprint accepts exactly the certificate with this SHA-256 and no other — the leaf alone, its
 // name and chain deliberately not consulted, as mesh-controller's own pin does (jetstream.go:116-132).
-func pinnedToFingerprint(want string) *tls.Config {
+// PinnedToFingerprint is public so that a process dialling the bus on its own terms, and the controller,
+// share this one pin (novox/hq issue 1633).
+func PinnedToFingerprint(want string) *tls.Config {
 	return &tls.Config{
 		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
 		MinVersion:         tls.VersionTLS12,

go/process/follow.go

@@ -0,0 +1,53 @@
+package process
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"git.novox.be/novox/mesh-sdk/go/membership"
+)
+
+// FollowMembership reads this account's own membership and follows it (novox/hq ADR 0160): apply is called
+// with every membership the mesh issues this account's node and module, the one already issued first and
+// each later one as it is published, so a change reaches a running process without a restart. A body that is
+// not a membership is handed to onBad (nil to ignore it) and skipped.
+//
+// The subject is followed before it is read, so an issue landing between the two is applied, not missed. The
+// read is the subject-addressed direct get, the one request this account may make of the stream; no
+// membership issued yet, or the read failing, is not an error: the process follows the subject and is told
+// when one arrives. The returned error is a refusal to follow at all.
+func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(error)) error {
+	if c.account.Node == "" || c.account.Module == "" {
+		return fmt.Errorf("the account names no node or module, so it has no membership")
+	}
+	bad := func(err error) {
+		if onBad != nil {
+			onBad(err)
+		}
+	}
+	take := func(body []byte) {
+		var issued membership.Membership
+		if err := json.Unmarshal(body, &issued); err != nil {
+			bad(fmt.Errorf("a membership arrived that is not one: %w", err))
+			return
+		}
+		apply(issued)
+	}
+	subject := membership.Subject(c.account.Node, c.account.Module)
+	if _, err := c.nc.Subscribe(subject, func(m *nats.Msg) { take(m.Data) }); err != nil {
+		return fmt.Errorf("cannot follow %s: %w", subject, err)
+	}
+	got, err := c.nc.Request(membership.DirectGetSubject(c.account.Node, c.account.Module), nil, 5*time.Second)
+	switch {
+	case err != nil:
+		bad(fmt.Errorf("cannot read the membership issued on %s yet (%v); following it", subject, err))
+	case got.Header.Get("Status") != "" || len(got.Data) == 0:
+		// None issued yet: the follow above delivers it.
+	default:
+		take(got.Data)
+	}
+	return nil
+}

go/process/follow_integration_test.go

@@ -0,0 +1,45 @@
+//go:build integration
+
+package process
+
+import (
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/membership"
+)
+
+// A process connected pinned to a test bus hears the membership published for its node and module, and
+// ignores what is not one (novox/hq issue 1633, AC2).
+func TestFollowMembershipHearsWhatIsIssuedToTheAccount(t *testing.T) {
+	url, fingerprint := selfSignedServer(t)
+	account := mailSenderAccount(t, url, fingerprint)
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	got := make(chan membership.Membership, 1)
+	bad := make(chan error, 4)
+	if err := conn.FollowMembership(func(m membership.Membership) { got <- m }, func(e error) { bad <- e }); err != nil {
+		t.Fatal(err)
+	}
+	if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte("not json")); err != nil {
+		t.Fatal(err)
+	}
+	if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte(`{"node":"novox","module":"mesh-mail","mesh":["10.0.0.1"]}`)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case m := <-got:
+		if m.Module != "mesh-mail" || len(m.Mesh) != 1 {
+			t.Fatalf("membership: %+v", m)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("no membership heard")
+	}
+	if len(bad) == 0 {
+		t.Error("the body that is not a membership was not reported")
+	}
+}

go/process/pin_test.go

@@ -0,0 +1,23 @@
+package process
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"testing"
+)
+
+// The pin accepts exactly the leaf certificate with the named SHA-256, and no other (novox/hq issue 1633).
+func TestThePinAcceptsOnlyTheNamedCertificate(t *testing.T) {
+	leaf := []byte("a certificate")
+	sum := sha256.Sum256(leaf)
+	cfg := PinnedToFingerprint("sha256:" + hex.EncodeToString(sum[:]))
+	if err := cfg.VerifyPeerCertificate([][]byte{leaf}, nil); err != nil {
+		t.Errorf("the pinned certificate was refused: %v", err)
+	}
+	if err := cfg.VerifyPeerCertificate([][]byte{[]byte("another")}, nil); err == nil {
+		t.Error("another certificate was accepted")
+	}
+	if err := cfg.VerifyPeerCertificate(nil, nil); err == nil {
+		t.Error("no certificate was accepted")
+	}
+}