novox/mesh-sdk
18
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

The lease write's contract: the seat mesh-grants, the lease cut, processes follow their membership from state alone (issue 774) #156

Merged
mesh-tools merged 9 commits from feat/l3-removal into main 2026-11-02 00:35:20 +00:00
Conversation 1 Commits 9 Files Changed 20
+72 -34
4 changed files with 72 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a4ed4e6742 - Show all commits
go/buscompose/buscompose_test.go
+16 -2
View File
@@ -161,8 +161,11 @@ func grantsAsTheControllers(t *testing.T, got, want string) {
intended := beyondTheControllers(user)
for g := range was {
// A runtime's direct get of its membership from the assignments stream went with the stream (research 072
// row 33): it reads the grants part's state memberships instead (ADR 0571 clause 7).
if !now[g] && !strings.HasPrefix(g, "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment."+strings.SplitN(user, ".", 2)[0]+".") {
// row 33): it reads the grants part's state memberships instead (ADR 0571 clause 7). And it no longer hears
// the subject, where the grants part stops publishing at the lease write (ADR 0571 clause 10, issue 774).
node := strings.SplitN(user, ".", 2)[0]
if !now[g] && !strings.HasPrefix(g, "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment."+node+".") &&
!strings.HasPrefix(g, "mesh.assignment."+node+".") {
t.Errorf("%s: the copy does not grant %s, which the controller's does", user, g)
}
}
@@ -880,6 +883,17 @@ func healthChecks(t *testing.T) {
if u.Kind == KindModule && MayPublish(perms, "$KV.mesh-delivery-assigner_policies.x") {
t.Errorf("%s writes the assigner's policies", u.Username())
}
// ADR 0571 clause 10 (issue 774, 774/S5/AC2): nobody hears memberships on mesh.assignment; they are read from
// the grants part's state.
if MaySubscribe(perms, "mesh.assignment.a.restic") {
t.Errorf("%s hears mesh.assignment", u.Username())
}
}
// Nor the runtime carrying the node's modules, which reads each one's membership from the state.
if perms, err := PermissionsFor(Principal{Kind: KindNodeTools, Node: "a", Module: RuntimeModule,
Carries: []Declared{{Module: "restic"}}}); err != nil || MaySubscribe(perms, "mesh.assignment.a.restic") ||
!MayPublish(perms, "$JS.API.DIRECT.GET.KV_mesh-grants_memberships.$KV.mesh-grants_memberships.a.restic") {
t.Errorf("the runtime hears mesh.assignment or reads no membership: %v", err)
}
// 1953 S-age (ADR 0565 §10): a log that says an age keeps its entries that long and no longer; one that says none, as
// long as the log (ADR 0297).
go/buscompose/nats.go
+8 -8
View File
@@ -1075,10 +1075,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
}
}
sub = append(sub, announcing(announced...)...)
// Its own membership (ADR 0160): the one subject a runtime derives for itself, heard live. Nothing else's.
sub = append(sub, MembershipSubject(p.Node, p.Module))
// And from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7): its one read of it,
// since the assignments stream's direct get went with the stream (research 072 row 33).
// Its own membership (ADR 0160), from the grants part's state memberships, its own key (novox/hq ADR 0571 clause
// 7): its one read of it, since the assignments stream's direct get went with the stream (research 072 row 33),
// and no longer heard on its subject, where the grants part stops publishing at the lease write (ADR 0571 clause
// 10, issue 774).
pub = append(pub, readGrants(GrantsMembershipsBucket, MembershipKey(p.Node, p.Module))...)
// 1b. The tools it calls, if its manifest says it calls any (novox/hq ADR 0152). The same
@@ -1240,10 +1240,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
}
}
}
// Every assigned module's membership on this node (ADR 0160): one per module, heard live and read from the
// grants part's state memberships (ADR 0571 clause 7). This node's and no other's — the one token
// that varies is the module, so the pattern is the machine's own assignments.
sub = append(sub, "mesh.assignment."+p.Node+".*")
// Every assigned module's membership on this node (ADR 0160): one per module, read and watched from the
// grants part's state memberships (ADR 0571 clause 7), and no longer heard on `mesh.assignment`, where the grants
// part stops publishing at the lease write (ADR 0571 clause 10, issue 774). This node's and no other's — the
// one token that varies is the module, so the pattern is the machine's own assignments.
pub = append(pub, readGrants(GrantsMembershipsBucket, p.Node+".*")...)
// And that it is there (novox/hq to-be 45 §3, S11): its own heartbeat, under its machine's
// name and no other's, on core NATS like the host's.
go/process/follow.go
+7 -15
View File
@@ -5,7 +5,6 @@ import (
"fmt"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/membership"
@@ -17,10 +16,10 @@ import (
// not a membership is handed to onBad (nil to ignore it) and skipped.
//
// It reads the membership from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7),
// watched so the value there now comes first and each change after it; and it hears the subject the grants part
// publishes on until the lease write. The assignments stream is no longer read: the bus module's tools delete it
// at the lease write (research 072 row 33). A state not there yet, or not readable, is told to onBad and the
// subject is followed alone. The returned error is a refusal to follow at all.
// watched so the value there now comes first and each change after it. Nothing else: the subject the grants part
// published on is no longer heard (ADR 0571 clause 10, issue 774), since it stops publishing at the lease write, and
// the assignments stream is no longer read (research 072 row 33). A state not there or not readable is a refusal to
// follow, returned: there is nothing else to follow.
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")
@@ -38,27 +37,20 @@ func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(er
}
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)
}
key := membership.GrantsKey(c.account.Node, c.account.Module)
js, err := jetstream.New(c.nc)
if err != nil {
bad(fmt.Errorf("cannot read %s of %s (%v); following %s alone", key, membership.GrantsBucket, err, subject))
return nil
return fmt.Errorf("cannot read %s of %s: %w", key, membership.GrantsBucket, err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
kv, err := js.KeyValue(ctx, membership.GrantsBucket)
if err != nil {
bad(fmt.Errorf("cannot read %s of %s yet (%v); following %s alone", key, membership.GrantsBucket, err, subject))
return nil
return fmt.Errorf("cannot read %s of %s: %w", key, membership.GrantsBucket, err)
}
w, err := kv.Watch(context.Background(), key)
if err != nil {
bad(fmt.Errorf("cannot watch %s of %s (%v); following %s alone", key, membership.GrantsBucket, err, subject))
return nil
return fmt.Errorf("cannot watch %s of %s: %w", key, membership.GrantsBucket, err)
}
go func() {
for e := range w.Updates() {
go/process/follow_integration_test.go
+41 -9
View File
@@ -3,14 +3,20 @@
package process
import (
"context"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"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).
// A process connected pinned to a test bus follows the membership the grants part writes for its node and module in
// its state memberships, and reports what is not one (novox/hq issue 1633, AC2); it no longer hears mesh.assignment,
// where the grants part stops publishing at the lease write (ADR 0571 clause 10; issue 774, 774/S5/AC1). Without the
// state it refuses to follow, naming it.
func TestFollowMembershipHearsWhatIsIssuedToTheAccount(t *testing.T) {
url, fingerprint := selfSignedServer(t)
account := mailSenderAccount(t, url, fingerprint)
@@ -20,26 +26,52 @@ func TestFollowMembershipHearsWhatIsIssuedToTheAccount(t *testing.T) {
}
defer conn.Close()
got := make(chan membership.Membership, 1)
got := make(chan membership.Membership, 4)
bad := make(chan error, 4)
if err := conn.FollowMembership(func(m membership.Membership) { got <- m }, func(e error) { bad <- e }); err != nil {
follow := func() error {
return conn.FollowMembership(func(m membership.Membership) { got <- m }, func(e error) { bad <- e })
}
if err := follow(); err == nil || !strings.Contains(err.Error(), membership.GrantsBucket) {
t.Fatalf("followed with no state memberships: %v", err)
}
js, err := jetstream.New(conn.nc)
if err != nil {
t.Fatal(err)
}
if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte("not json")); err != nil {
ctx := context.Background()
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: membership.GrantsBucket})
if 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 {
key := membership.GrantsKey("novox", "mesh-mail")
if _, err := kv.Put(ctx, key, []byte("not json")); err != nil {
t.Fatal(err)
}
if err := follow(); err != nil {
t.Fatal(err)
}
// Published on the subject, as the grants part did until the write: not heard.
if err := conn.nc.Publish(membership.Subject("novox", "mesh-mail"), []byte(`{"node":"novox","module":"mesh-mail","mesh":["10.0.0.9"]}`)); err != nil {
t.Fatal(err)
}
if _, err := kv.Put(ctx, key, membership.EncodeGrantsMembership(membership.Membership{Node: "novox", Module: "mesh-mail",
Mesh: []string{"10.0.0.1"}})); err != nil {
t.Fatal(err)
}
select {
case m := <-got:
if m.Module != "mesh-mail" || len(m.Mesh) != 1 {
if m.Module != "mesh-mail" || len(m.Mesh) != 1 || m.Mesh[0] != "10.0.0.1" {
t.Fatalf("membership: %+v", m)
}
case <-time.After(5 * time.Second):
t.Fatal("no membership heard")
t.Fatal("no membership followed")
}
select {
case m := <-got:
t.Errorf("a second membership was applied, from the subject: %+v", m)
case <-time.After(300 * time.Millisecond):
}
if len(bad) == 0 {
t.Error("the body that is not a membership was not reported")
t.Error("the value that is not a membership was not reported")
}
}