novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Read each membership's last value from mesh-grants' own state and say it on its subject without the assignments stream, so memberships keep flowing once nats_stream_remove takes the stream at the lease write (issue 774, research 072 row 33)

Browse Source
This commit is contained in:
jschoubben
2026-10-31 10:25:20 +01:00
parent 4b194215dc
commit e7f51caedb
5 changed files with 76 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/cmd/mesh-grants/main.go
+3 -3
View File
@@ -4,8 +4,8 @@
// list and each manifest's buckets, logs and streams with the bus composer the controller uses (mesh-sdk buscompose).
//
// **Memberships are published for real** (the operator, 2026-10-25 22:31): since release 6 nobody publishes them, so a
// module placed since cannot serve its seat or reach a new state. Each is published on the assignments stream only when
// it differs from the one last published there. **A bucket, log or stream a manifest declares that the bus lacks is
// module placed since cannot serve its seat or reach a new state. Each is published in the part's own state memberships,
// and said on its subject, only when it differs from the one last published there. **A bucket, log or stream a manifest declares that the bus lacks is
// created** (the operator, 2026-10-26 08:54; issue 1790), once per manifest, with the manifest's limits, on the event
// that brought the manifest; one the bus holds is never changed. **The user list is written sealed to the bus node**
// (ADR 0547 clause 7) as an object of the part's own bucket named in `user-list.sealed`, which the bus node follows only
@@ -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}, Consumers: creating{ctx: ctx, js: js},
Publish: memberships{nc: nc}, 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
modules/mesh-grants/cmd/mesh-grants/publish.go
+8 -39
View File
@@ -1,55 +1,24 @@
package main
import (
"context"
"errors"
"fmt"
"strconv"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/membership"
)
// memberships publishes each membership on the assignments stream, last per subject, over the one last read there
// (ExpectLastSequencePerSubject, so a membership another writer published meanwhile is not overwritten, issues 930 and
// 934), and reads the last one by a direct get of its subject. The one place the part publishes on the bus.
// memberships says each membership the part wrote to its state on its subject, `mesh.assignment.<node>.<module>`, as a
// plain message: a notice for a runtime that follows the subject alone, which the assignments stream also keeps while it
// exists. It reads nothing there and asks no stream for an acknowledgement, so the stream's removal at the lease write
// (research 072 row 33) stops no membership (issue 774). The one place the part publishes a membership on the bus.
type memberships struct {
ctx context.Context
nc *nats.Conn
js jetstream.JetStream
nc *nats.Conn
}
func (m memberships) LastMembership(node, module string) ([]byte, uint64, bool, error) {
ctx, cancel := context.WithTimeout(m.ctx, requestWithin)
defer cancel()
msg, err := m.nc.RequestWithContext(ctx, membership.DirectGetSubject(node, module), nil)
if err != nil {
if errors.Is(err, context.DeadlineExceeded) {
err = fmt.Errorf("no answer in %s: a read the bus refuses is answered by none (%w)", requestWithin, err)
}
return nil, 0, false, err
}
switch status := msg.Header.Get("Status"); status {
case "":
case "404":
return nil, 0, false, nil
default:
return nil, 0, false, fmt.Errorf("the assignments stream answered %s %s", status, msg.Header.Get("Description"))
}
seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64)
if err != nil {
return nil, 0, false, fmt.Errorf("the membership of %s on %s carries no sequence", module, node)
}
return msg.Data, seq, true, nil
}
func (m memberships) PublishMembership(node, module string, body []byte, after uint64) error {
ctx, cancel := context.WithTimeout(m.ctx, requestWithin)
defer cancel()
if _, err := m.js.Publish(ctx, membership.Subject(node, module), body, jetstream.WithExpectLastSequencePerSubject(after)); err != nil {
return fmt.Errorf("the membership of %s on %s after sequence %d: %w", module, node, after, err)
func (m memberships) PublishMembership(node, module string, body []byte) error {
if err := m.nc.Publish(membership.Subject(node, module), body); err != nil {
return fmt.Errorf("the membership of %s on %s: %w", module, node, err)
}
return nil
}
modules/mesh-grants/grants_test.go
+22 -33
View File
@@ -47,10 +47,8 @@ type fakeBus struct {
calls []string
events []string
writes []string
// The assignments stream: each subject's last membership and sequence, every publish, and whether it is refused.
// Each membership's subject: the last membership said there, every notice, and whether it is refused.
stream map[string][]byte
seqs map[string]uint64
seq uint64
publishes []string
refusePub bool
// hold, when set, keeps every nats_users call waiting until it is closed: a pass waiting on the bus.
@@ -106,29 +104,17 @@ func (b *fakeBus) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) er
return nil
}
func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) {
b.mu.Lock()
defer b.mu.Unlock()
subject := "mesh.assignment." + node + "." + module
v, ok := b.stream[subject]
return v, b.seqs[subject], ok, nil
}
func (b *fakeBus) PublishMembership(node, module string, body []byte, after uint64) error {
func (b *fakeBus) PublishMembership(node, module string, body []byte) error {
b.mu.Lock()
defer b.mu.Unlock()
subject := "mesh.assignment." + node + "." + module
if b.refusePub {
return errors.New("Permissions Violation for Publish to " + subject)
}
if b.seqs[subject] != after {
return fmt.Errorf("wrong last sequence for %s: %d", subject, b.seqs[subject])
}
if b.stream == nil {
b.stream, b.seqs = map[string][]byte{}, map[string]uint64{}
b.stream = map[string][]byte{}
}
b.seq++
b.stream[subject], b.seqs[subject] = body, b.seq
b.stream[subject] = body
b.publishes = append(b.publishes, subject)
return nil
}
@@ -1773,10 +1759,11 @@ func TestGrants(t *testing.T) {
b.manifestOf("mod-x", strings.Replace(xManifest, `"own-secrets"`, `"claims":`+row.claims+`,"own-secrets"`, 1))
}
b.loaded = shadowOf(t, b)
// The last membership is the part's own state's, never the assignments stream's: the stream goes at the lease
// write (research 072 row 33), and a part reading its last there would publish nothing after it.
if row.last != nil {
raw, _ := json.Marshal(row.last(composed(b)))
b.stream = map[string][]byte{"mesh.assignment." + key: raw}
b.seqs = map[string]uint64{"mesh.assignment." + key: 4}
b.own[grants.StateMemberships] = map[string][]byte{key: raw}
}
res := partOn(b).Pass()
if got := slices.Contains(res.Published, key); got != row.published {
@@ -1784,12 +1771,12 @@ func TestGrants(t *testing.T) {
}
// The state holds the bare membership the assignments stream carries, through mesh-sdk's codec, and
// only one published (issue 774); what became of it is the kept pass's, which grants_membership reads.
state, _, inState, _ := b.ownGet(grants.StateMemberships, key)
state, _, _, _ := b.ownGet(grants.StateMemberships, key)
ms, _ := partOn(b).MembershipOf("a", "mod-x")
pass := fmt.Sprint(ms.(map[string]any)["last_pass"])
if row.published {
if row.names == "" && row.last != nil && !strings.Contains(string(b.stream["mesh.assignment."+key]), "10.0.0.1") {
t.Errorf("the private network of the last membership is not carried: %s", b.stream["mesh.assignment."+key])
if row.names == "" && row.last != nil && !strings.Contains(string(state), "10.0.0.1") {
t.Errorf("the private network of the last membership is not carried: %s", state)
}
got, err := membership.DecodeGrantsMembership(state)
var sent membership.Membership
@@ -1802,7 +1789,7 @@ func TestGrants(t *testing.T) {
}
return
}
if inState {
if was, _ := json.Marshal(row.last(composed(b))); !bytes.Equal(state, was) {
t.Errorf("a membership held back reached the state: %s", state)
}
said := strings.Join(res.NotPublished, "\n")
@@ -1814,18 +1801,21 @@ func TestGrants(t *testing.T) {
}
})
}
t.Run("a refused publish: no exit, said, published at the next event", func(t *testing.T) {
// The assignments stream removed (nats_stream_remove at the lease write) and its subject refused: every membership
// is still written to the state, where each runtime reads it, and the refusal is said (issue 774).
t.Run("the stream gone and its subject refused: memberships still written, the refusal said", func(t *testing.T) {
b := fixture()
b.loaded = shadowOf(t, b)
b.refusePub = true
p := partOn(b)
refused := p.Compose()
if len(refused.Published) != 0 || len(refused.NotPublished) != 3 || !strings.Contains(refused.NotPublished[0], "Permissions Violation") {
t.Errorf("a refused publish = %+v: said, and nothing published", refused)
if !slices.Equal(refused.Published, []string{"a.mod-x", "a.nats", "b.mod-y"}) || refused.MembershipsWritten != 3 ||
!strings.Contains(strings.Join(refused.Errors, "\n"), "Permissions Violation") {
t.Errorf("with the stream gone = %+v: every membership written, the refusal said", refused)
}
b.refusePub = false
if next := p.Compose(); !slices.Equal(next.Published, []string{"a.mod-x", "a.nats", "b.mod-y"}) {
t.Errorf("the next event published %v", next.Published)
if next := p.Compose(); len(next.Published) != 0 || next.MembershipsWritten != 0 {
t.Errorf("the next event published %v again", next.Published)
}
})
t.Run("a tool answers while a pass waits on the bus", func(t *testing.T) {
@@ -1974,16 +1964,15 @@ func TestGrants(t *testing.T) {
b := fixture()
b.loaded = shadowOf(t, b)
recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{}))
b.stream, b.seqs = map[string][]byte{}, map[string]uint64{}
for i, key := range []string{"a.mod-x", "b.mod-y"} {
b.own[grants.StateMemberships] = map[string][]byte{}
for _, key := range []string{"a.mod-x", "b.mod-y"} {
raw, _ := json.Marshal(grants.Memberships(recs)[key].Membership)
var m map[string]any
_ = json.Unmarshal(raw, &m)
seat, verb := brokerSeat()
m["seats"] = []any{map[string]any{"seat": seat, "verb": verb, "subject": "mesh.seat." + seat + ".tool." + verb}}
raw, _ = json.Marshal(m)
subject := "mesh.assignment." + key
b.stream[subject], b.seqs[subject] = raw, uint64(i+1)
b.own[grants.StateMemberships][key] = raw
}
var logged []string
var lmu sync.Mutex
modules/mesh-grants/internal/grants/memberships.go
+15 -18
View File
@@ -4,7 +4,6 @@ import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"reflect"
"slices"
"sort"
@@ -25,12 +24,12 @@ var MembershipGaps = []Gap{
{Input: "membership: root-free machines", Why: "asked of the controller's root-free verb at each pass and kept for that pass alone; a machine it does not call free (an error included) is composed as not free, and the pass says why"},
}
// Publisher publishes a membership on the assignments stream (membership.Subject), last per subject, and reads the one
// last published there. A refusal (the grant not yet on the bus) is an error the part says and tries again at the next
// event.
// Publisher says a membership the part wrote to its state memberships on its subject (membership.Subject), as a plain
// message, for a runtime that follows the subject alone. The state is where a membership is published and read back
// (ADR 0571 clause 7); the subject is a notice that depends on no stream, so the assignments stream's removal at the
// lease write (research 072 row 33) stops nothing. A refusal is said and the membership stays written.
type Publisher interface {
LastMembership(node, module string) (body []byte, seq uint64, found bool, err error)
PublishMembership(node, module string, body []byte, after uint64) error
PublishMembership(node, module string, body []byte) error
}
// Membership is one module's membership on one node, as composed now, and what became of its publishing.
@@ -73,18 +72,20 @@ const (
MembershipUnchanged = "unchanged"
)
// publish publishes m when it differs from the membership last published for its assignment, carrying that one's private
// network and received contributions; it answers m as published (or why not) and whether it was published now. held is
// every holding the memberships composed now carry (Holdings.ByMemberships).
func publish(p Publisher, m Membership, h Holdings, held map[string]bool) (Membership, bool) {
// publish says whether m is to be published: whether it differs from was, the membership last published (the value of
// its key in the part's own state memberships, found when it is there), carrying that one's private network and
// received contributions. It answers m as composed and published, or why not. held is every holding the memberships
// composed now carry (Holdings.ByMemberships). Pure: the caller writes and notifies.
func publish(p Publisher, was []byte, found bool, m Membership, h Holdings, held map[string]bool) (Membership, bool) {
if p == nil {
m.Published = "not published: no publisher here (only the part's process publishes)"
return m, false
}
was, seq, found, err := p.LastMembership(m.Node, m.Module)
if err != nil {
m.Published = "not published: the membership last published cannot be read: " + err.Error()
return m, false
if found {
// A value of the state is the bare membership, or the wrapper written before 2026-10-30 (mesh-sdk's one codec);
// one that is neither is no last membership, and m is a first one.
bare, err := membership.BareGrantsMembership(was)
was, found = bare, err == nil
}
if found {
var last membership.Membership
@@ -122,10 +123,6 @@ func publish(p Publisher, m Membership, h Holdings, held map[string]bool) (Membe
}
}
}
if err := p.PublishMembership(m.Node, m.Module, body, seq); err != nil {
m.Published = fmt.Sprintf("not published (tried again at the next event): %v", err)
return m, false
}
m.Published = MembershipPublished
return m, true
}
modules/mesh-grants/internal/grants/part.go
+28 -10
View File
@@ -140,7 +140,7 @@ type Result struct {
}
// NothingTakesEffect is what every composition says it did on the bus beyond its own state and the memberships.
const NothingTakesEffect = "memberships (on the assignments stream, unless one would leave a seat a placed module claims with no holder, or drop a capability its holder still asks for, which stays a shadow difference), and each bucket, log or stream a manifest declares that the bus lacks, created once with the manifest's limits (the operator, 2026-10-26 08:54); the user list, the grants and any change to a stream the bus holds stay in shadow until take-over"
const NothingTakesEffect = "memberships (in its own state memberships, each said on its subject, unless one would leave a seat a placed module claims with no holder, or drop a capability its holder still asks for, which stays a shadow difference), and each bucket, log or stream a manifest declares that the bus lacks, created once with the manifest's limits (the operator, 2026-10-26 08:54); the user list, the grants and any change to a stream the bus holds stay in shadow until take-over"
// Part is the grants part: it reads, calls and writes through these, and nothing else.
type Part struct {
@@ -328,7 +328,33 @@ func (p *Part) Compose() Result {
}
seated := holdings.ByMemberships(members...)
for _, key := range sortedKeys(ms) {
m, now := publish(p.Publish, ms[key], holdings, seated)
// The last membership is the state's own (ADR 0571 clause 7), never the assignments stream's, which the bus
// module's tools remove at the lease write (research 072 row 33): a last read there would freeze every membership
// once it is gone.
was, _, found, err := p.Store.Get(StateMemberships, key)
if err != nil {
res.NotPublished = append(res.NotPublished, key+": not published: the membership last published cannot be read from its own "+StateMemberships+": "+err.Error())
continue
}
m, now := publish(p.Publish, was, found, ms[key], holdings, seated)
// Published is written to the state, where every runtime reads it (process.FollowMembership); only one
// published, so a membership held back never reaches a reader. The subject is a notice after it.
if now {
body := membership.EncodeGrantsMembership(m.Membership)
if _, err := p.Store.Put(StateMemberships, key, body); err != nil {
res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be written: %v", StateMemberships, key, err))
m.Published, now = "not published (tried again at the next event): its own "+StateMemberships+" cannot be written", false
} else {
res.MembershipsWritten++
if err := p.Publish.PublishMembership(m.Node, m.Module, body); err != nil {
res.Errors = append(res.Errors, fmt.Sprintf("the membership of %s is written, its notice on its subject not said: %v", key, err))
}
}
}
// One unchanged is written bare once where the state still holds the wrapper written before 2026-10-30.
if m.Published == MembershipUnchanged && p.changed(&res, StateMemberships, key, membership.EncodeGrantsMembership(m.Membership)) {
res.MembershipsWritten++
}
switch {
case now:
res.Published = append(res.Published, key)
@@ -342,14 +368,6 @@ func (p *Part) Compose() Result {
}
res.Retired[key] = m.Retired
}
// The state holds the bare membership as the assignments stream carries it, through mesh-sdk's one codec (issue
// 774): only one published now or unchanged, so a membership held back never reaches a reader by this way. What
// became of each at this pass is the pass's own (Published, NotPublished, Retired), kept with it (Pass).
if now || m.Published == MembershipUnchanged {
if p.changed(&res, StateMemberships, key, membership.EncodeGrantsMembership(m.Membership)) {
res.MembershipsWritten++
}
}
}
held, heldErr := HeldStreams(p.Call)
var ask Ask