Issue every node's memberships when it is composed, so a module taken in with a new state is not refused it (issue 249 met again, issue 1945)

Release 6 removed the push (788fb681), and with it the only writer of the
memberships on ASSIGNMENTS. A module taken in since kept the membership the
old push last issued: mesh-delivery-gate, taken in with the state
"conditions" at 20:33 on 2026-10-25, was granted its bucket on the bus and
refused it by novox's tool runner every second, which keeps a module to its
membership (node-tools internal/bus/state.go issuedState).

A node's composition pass now issues the membership of each module assigned
to it, from the same records the grants are composed from, with the node's
private-network addresses and received contributions, before its
compositions are written. Only one that differs from the one last issued is
published, and only over that one (issues 930, 934). The memberships' stream,
every declared bucket and log are asserted first; the root-free judgement
runs only when a held seat promises a verified sender.

NATS-native piece moved (research 072, 03-domains, mesh-grants): the
membership is issued again by the take-in's pass, from the grants' records,
the step before mesh-grants takes both over.

cmd/mesh-controller/composed_live.go

@@ -88,6 +88,8 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
 		}
 		return u.Policy(), u.From, u.Why
 	}
+	// The memberships of each node's modules are issued by its composition (memberships.go): read once for the pass.
+	issuing, issuingErr := membershipsFor(ctx, open, link.OverNATS{Conn: bus.Conn(), JS: bus.Context()})
 	sort.Strings(nodes)
 	out := map[string][]string{}
 	errsOf := map[string][]error{}
@@ -115,6 +117,13 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
 				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
 				return
 			}
+			// Its modules' memberships, before its compositions (issue 249): a failure is the node's, and its compositions
+			// are still written; the take-in that asked tries the node again.
+			if issuingErr != nil {
+				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, issuingErr))
+			} else if err := issuing.issue(ctx, node, declared); err != nil {
+				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
+			}
 			// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
 			if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
 				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))

cmd/mesh-controller/memberships.go

@@ -0,0 +1,186 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"reflect"
+	"slices"
+	"sort"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"github.com/nats-io/nats.go"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/link"
+)
+
+// **A node's composition issues its modules' memberships** (novox/hq ADR 0160, ADR 0201; issue 249 met again under
+// issue 1945).
+//
+// A membership is what an assignment serves and reaches: its subjects, its seats, and the state, logs, streams and waits
+// its manifest declares, published last-per-subject on the AssignmentsStream, where the tool runner and every bus process
+// read and follow their own. A push issued them before it sent a declaration; release 6 removed the push and nothing
+// issued them after, so a module taken in with a new state, read or stream was granted it on the bus and refused it by its
+// runtime, which keeps a module to the membership last issued (2026-10-25: mesh-delivery-gate's `conditions`, on novox
+// from 20:33). Composing a node is now where its memberships are issued: from the same records the bus's grants are
+// composed from (so what a runtime serves and what its account may are one composition), with the private network's
+// addresses and what each module receives from the node's composition, before its compositions are written (the code
+// arriving first was refused its own state, issue 249).
+//
+// Only a membership that differs from the one last issued is published, and only over that one, so a pass that changes
+// nothing re-serves nothing and a newer one issued meanwhile is not overwritten (issues 930, 934).
+
+// membershipPass is what one composer pass issues every node's memberships from, read once for the pass.
+type membershipPass struct {
+	records broker.Records
+	where   broker.Placements
+	bus     link.OverNATS
+}
+
+// membershipsFor reads the records, asserts the memberships' stream and every declared bucket and log (a membership
+// names them, and a module whose state does not exist fails its first read), and judges which machines are root-free
+// (ADR 0259 §8: a channel's verified sender is composed only from one) when a seat promises one, as a push did.
+func membershipsFor(ctx context.Context, open *stores, bus link.OverNATS) (*membershipPass, error) {
+	records, err := open.inventory.BusRecords(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("the memberships are not issued: %w", err)
+	}
+	raiser := broker.OnConn(bus.Conn)
+	for _, s := range broker.MeshStreams() {
+		if s.Name != broker.AssignmentsStream {
+			continue
+		}
+		if err := raiser.EnsureStream(s); err != nil {
+			return nil, fmt.Errorf("the memberships are not issued: their stream: %w", err)
+		}
+	}
+	buckets, err := open.inventory.DeclaredBuckets(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("the memberships are not issued: the modules' state: %w", err)
+	}
+	if _, err := broker.RaiseBuckets(raiser, buckets); err != nil {
+		return nil, fmt.Errorf("the memberships are not issued: the modules' state could not be asserted: %w", err)
+	}
+	logs, err := open.inventory.DeclaredLogs(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("the memberships are not issued: the modules' logs: %w", err)
+	}
+	if _, err := broker.RaiseLogs(raiser, logs); err != nil {
+		return nil, fmt.Errorf("the memberships are not issued: the modules' logs could not be asserted: %w", err)
+	}
+	var whys map[string]string
+	if promisesVerifiedSender(records) {
+		records.RootFree, whys = rootVerdictsOf(judgeRootFree(ctx, newRootReader(ctx, open.inventory, bus.Conn),
+			records.Nodes, time.Now()))
+	}
+	where := broker.PlacementsOf(records, records.Interchangeable)
+	for _, line := range droppedVerifiedSenders(records, where, whys) {
+		fmt.Println(line)
+	}
+	return &membershipPass{records: records, where: where, bus: bus}, nil
+}
+
+// promisesVerifiedSender says a seat some module holds promises a verified sender: only then is the root-free judgement,
+// a discovery on the bus, worth its seconds. Pure.
+func promisesVerifiedSender(records broker.Records) bool {
+	for _, declared := range records.Assigned {
+		for _, d := range declared {
+			for _, s := range d.Holds {
+				if slices.Contains(s.Capabilities, "verified-sender") {
+					return true
+				}
+			}
+		}
+	}
+	return false
+}
+
+// issue issues node's memberships on the bus, from its composition's addresses and received contributions.
+func (p *membershipPass) issue(ctx context.Context, node string, declared sendable) error {
+	issued, err := issueComposedMemberships(node, p.records, p.where, declared.Mesh, declared.Received,
+		func(node, module string) ([]byte, uint64, error) {
+			m, err := p.bus.JS.GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject(node, module), nats.Context(ctx))
+			if err != nil {
+				return nil, 0, err
+			}
+			return m.Data, m.Sequence, nil
+		},
+		func(node, module string, body []byte, seq uint64) error {
+			return p.bus.PublishMembershipAfter(ctx, node, module, body, seq)
+		})
+	if len(issued) > 0 {
+		fmt.Printf("memberships: %s: issued %s\n", node, strings.Join(issued, ", "))
+	}
+	return err
+}
+
+// issueComposedMemberships composes the membership of every module assigned to node and publishes each one that differs
+// from the one last issued, over that one (seq 0 when none was: only if none was issued meanwhile). It answers the
+// modules it issued; every failure is tried past and joined in the error.
+func issueComposedMemberships(node string, records broker.Records, where broker.Placements, mesh []string,
+	received map[string]map[string][]catalogue.Contribution,
+	last func(node, module string) ([]byte, uint64, error),
+	publish func(node, module string, body []byte, seq uint64) error) ([]string, error) {
+	var issued []string
+	var errs []error
+	for _, d := range records.Assigned[node] {
+		body, err := composedMembership(node, d, where, mesh, received[d.Module])
+		if err != nil {
+			errs = append(errs, fmt.Errorf("%s on %s: its membership: %w", d.Module, node, err))
+			continue
+		}
+		was, seq, err := last(node, d.Module)
+		switch {
+		case errors.Is(err, nats.ErrMsgNotFound):
+			was, seq = nil, 0
+		case err != nil:
+			errs = append(errs, fmt.Errorf("%s on %s: its membership as last issued could not be read: %w", d.Module, node, err))
+			continue
+		}
+		if sameMembership(was, body) {
+			continue
+		}
+		if err := publish(node, d.Module, body, seq); err != nil {
+			errs = append(errs, fmt.Errorf("%s on %s: its membership could not be issued: %w", d.Module, node, err))
+			continue
+		}
+		issued = append(issued, d.Module)
+	}
+	sort.Strings(issued)
+	return issued, errors.Join(errs...)
+}
+
+// composedMembership is one assignment's membership as a node's composition issues it: MembershipFor, with the private
+// network's addresses and what the module receives, by requirement (ADR 0167).
+func composedMembership(node string, d broker.Declared, where broker.Placements, mesh []string,
+	received map[string][]catalogue.Contribution) ([]byte, error) {
+	m := broker.MembershipFor(node, d, where)
+	m.Mesh = mesh
+	for requirement, given := range received {
+		raw, err := json.Marshal(given)
+		if err != nil {
+			return nil, err
+		}
+		if m.Receives == nil {
+			m.Receives = map[string]json.RawMessage{}
+		}
+		m.Receives[requirement] = raw
+	}
+	return json.Marshal(m)
+}
+
+// sameMembership says two memberships say the same, whatever the order of their fields. Pure.
+func sameMembership(was, now []byte) bool {
+	if was == nil {
+		return false
+	}
+	var a, b any
+	if json.Unmarshal(was, &a) != nil || json.Unmarshal(now, &b) != nil {
+		return false
+	}
+	return reflect.DeepEqual(a, b)
+}

cmd/mesh-controller/reissue_verified_sender_test.go

@@ -108,14 +108,83 @@ func TestADroppedVerifiedSenderIsSaidWithItsReason(t *testing.T) {
 	}
 }
 
-func TestAMembershipNeverIssuedIsLeftToItsOwnSend(t *testing.T) {
-	records, _ := verifiedSenderRecords(map[string]bool{"novox": true})
-	bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
-	var said []string
-	n := reissueVerifiedSenders(records, broker.PlacementsOf(records, nil), bus.get, bus.put,
-		func(f string, a ...any) { said = append(said, f) })
-	if n != 0 || len(said) != 0 {
-		t.Fatalf("issued %d, said %v", n, said)
+// **A node's composition issues every membership that changed** (issue 249 met again under issue 1945): release 6 removed
+// the push that issued memberships, and mesh-delivery-gate taken in with a new state `conditions` was granted its bucket and
+// refused it by the tool runner, which reads the membership last issued (2026-10-25, novox). The self-check's reissue
+// leaves one never issued alone: the composition issues it.
+func TestAComposedNodeIssuesEveryMembershipThatChanged(t *testing.T) {
+	gate := func(states ...string) broker.Declared {
+		d := broker.Declared{Module: "mesh-delivery-gate"}
+		for _, s := range states {
+			d.State = append(d.State, broker.Bucket{Module: "mesh-delivery-gate", Name: s})
+		}
+		return d
+	}
+	records := func(d broker.Declared) broker.Records {
+		return broker.Records{Nodes: []string{"novox"}, Assigned: map[string][]broker.Declared{"novox": {d}}}
+	}
+	issuedAs := func(d broker.Declared, mesh []string) []byte {
+		body, err := composedMembership("novox", d, broker.PlacementsOf(records(d), nil), mesh, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return body
+	}
+	newer := []byte(`{"node":"novox","module":"mesh-delivery-gate","newer":true}`)
+	for _, c := range []struct {
+		name    string
+		before  []byte // the membership last issued, nil for none
+		between bool   // another issued meanwhile, after the read
+		issued  []string
+		states  []string // what the bus holds after, by state name; nil when it must hold newer
+		err     string
+	}{
+		{name: "never issued is issued", issued: []string{"mesh-delivery-gate"}, states: []string{"asks", "conditions"}},
+		{name: "a new state is issued", before: issuedAs(gate("asks"), []string{"10.0.0.1"}),
+			issued: []string{"mesh-delivery-gate"}, states: []string{"asks", "conditions"}},
+		{name: "unchanged is not issued again", before: issuedAs(gate("asks", "conditions"), []string{"10.0.0.1"}),
+			states: []string{"asks", "conditions"}},
+		{name: "a newer one meanwhile is not overwritten", before: issuedAs(gate("asks"), []string{"10.0.0.1"}),
+			between: true, err: "wrong last sequence"},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			r := records(gate("asks", "conditions"))
+			where := broker.PlacementsOf(r, nil)
+			bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
+			if c.before != nil {
+				bus.last["novox/mesh-delivery-gate"], bus.seq["novox/mesh-delivery-gate"] = c.before, 1
+			}
+			if c.between {
+				bus.between = func() { bus.last["novox/mesh-delivery-gate"], bus.seq["novox/mesh-delivery-gate"] = newer, 2 }
+			}
+			if n := reissueVerifiedSenders(r, where, bus.get, bus.put, func(string, ...any) {}); n != 0 {
+				t.Fatalf("the self-check's reissue issued %d", n)
+			}
+			issued, err := issueComposedMemberships("novox", r, where, []string{"10.0.0.1"}, nil, bus.get, bus.put)
+			if c.err == "" && err != nil || c.err != "" && (err == nil || !strings.Contains(err.Error(), c.err)) {
+				t.Fatalf("error %v, want %q", err, c.err)
+			}
+			if strings.Join(issued, ",") != strings.Join(c.issued, ",") {
+				t.Fatalf("issued %v, want %v", issued, c.issued)
+			}
+			if c.states == nil {
+				if string(bus.last["novox/mesh-delivery-gate"]) != string(newer) {
+					t.Fatalf("the newer membership was overwritten: %s", bus.last["novox/mesh-delivery-gate"])
+				}
+				return
+			}
+			var m broker.Membership
+			if err := json.Unmarshal(bus.last["novox/mesh-delivery-gate"], &m); err != nil {
+				t.Fatal(err)
+			}
+			var names []string
+			for _, s := range m.State {
+				names = append(names, s.Name)
+			}
+			if strings.Join(names, ",") != strings.Join(c.states, ",") || strings.Join(m.Mesh, ",") != "10.0.0.1" {
+				t.Fatalf("the bus holds states %v and mesh %v, want %v", names, m.Mesh, c.states)
+			}
+		})
 	}
 }