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

Stop declaring and reading the ASSIGNMENTS stream: a runtime reads its membership from the grants part's state, and the stream is deleted at the lease write and must not be made again (issue 774, research 072 row 33)

Browse Source
This commit is contained in:
jschoubben
2026-10-31 09:21:04 +01:00
parent 4c54aec972
commit 7a9c8570cb
7 changed files with 62 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/buscompose/buscompose_test.go
+10 -1
View File
@@ -15,6 +15,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/membership"
"git.novox.be/novox/mesh-sdk/go/provisions"
"github.com/nats-io/nats.go/jetstream"
)
@@ -67,6 +68,9 @@ func mayBeyondTheControllers(user string) map[string]bool {
key = parts[0] + ".*"
}
out := map[string]bool{}
if membership.GrantsBucket != GrantsMembershipsBucket || membership.GrantsKey("a", "b") != MembershipKey("a", "b") {
panic("the membership package's grants bucket does not agree with buscompose's") // FollowMembership reads it
}
for _, s := range readGrants(GrantsMembershipsBucket, key) {
out[s] = true
}
@@ -151,11 +155,16 @@ func grantsAsTheControllers(t *testing.T, got, want string) {
}
intended := beyondTheControllers(user)
for g := range was {
if !now[g] {
// 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]+".") {
t.Errorf("%s: the copy does not grant %s, which the controller's does", user, g)
}
}
for g := range now {
if strings.HasPrefix(g, "$JS.API.DIRECT.GET."+AssignmentsStream+".") && !strings.HasSuffix(user, "."+GrantsModule) {
t.Errorf("%s: the copy grants %s, a read of the assignments stream, which goes at the lease write", user, g)
}
if !was[g] && !intended[g] && !mayBeyondTheControllers(user)[g] && !createdAdvisory.MatchString(`"`+g+`", `) &&
!createsAnAccountsConsumer(user, g) && !engineBeyond(user, g) {
t.Errorf("%s: the copy grants %s beyond the controller's, and no decision names it", user, g)
go/buscompose/creates_streams_test.go
+2
View File
@@ -71,6 +71,8 @@ func TestTheGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_undeclared", false},
{GrantsModule, "$JS.API.STREAM.CREATE.EVENTS", false},
{GrantsModule, "$JS.API.STREAM.CREATE.ASSIGNMENTS", false},
// Research 072 row 33: a runtime reads its membership from the grants part's state, never the assignments stream.
{"mesh-delivery-builder", "$JS.API.DIRECT.GET.ASSIGNMENTS.mesh.assignment.novox.mesh-delivery-builder", false},
{GrantsModule, "$JS.API.STREAM.INFO.EVENTS", false},
{"mesh-delivery-builder", "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_steps", false},
// The node-engine's acts: created by the grants part, written by each node's engine under its own keys alone.
go/buscompose/nats.go
+5 -8
View File
@@ -1061,12 +1061,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, read
// directly from the stream and followed live. Nothing else's.
// 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))
pub = append(pub, "$JS.API.DIRECT.GET."+AssignmentsStream+"."+MembershipSubject(p.Node, p.Module))
// And from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7): the read that replaces
// the stream's at the lease write, when the grants part stops publishing there.
// 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).
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
@@ -1228,11 +1226,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
}
}
}
// Every assigned module's membership on this node (ADR 0160): one per module, read
// directly from the stream and followed live. This node's and no other's — the one token
// 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+".*")
pub = append(pub, "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment."+p.Node+".*")
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/buscompose/streams.go
+3 -8
View File
@@ -171,14 +171,9 @@ func MeshStreams() []Stream {
Why: "one declaration per node, always the newest; a node that sees sequence n refuses " +
"n-1 by construction (issue 107)",
},
{
Name: AssignmentsStream,
Subjects: []string{"mesh.assignment.*.*"},
Retention: RetentionLastPerSubject,
Direct: true,
Why: "one membership per assignment, always the newest: what the mesh issued this module " +
"on this machine to serve and to reach (ADR 0160); read directly by the runtime it is for",
},
// The assignments stream (ADR 0160) is no longer the mesh's: a runtime reads its membership from the grants
// part's state memberships (ADR 0571 clause 7), and the bus module's tools delete the stream at the lease write
// (research 072 row 33), so it is not declared here, which would make it again.
{
Name: EventsStream,
// A seat's own events ride here too: they are 1:many like any event, and the
go/catalogue/streams.go
+1 -1
View File
@@ -48,7 +48,7 @@ var ReservedStreamPrefixes = []string{"SEAT_", "LOG_", "KV_", "OBJ_", "DEAD_"}
// MeshStreamNames are the mesh's own streams by name. Written here as well as in the broker, which this
// package cannot import; a test keeps them agreeing.
var MeshStreamNames = []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS", "PROV_APPLIED", "CHECKS_READY", "CHECKS_OTHER"}
var MeshStreamNames = []string{"CONTROL", "NODES", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS", "PROV_APPLIED", "CHECKS_READY", "CHECKS_OTHER"}
// StreamDeclaration is one stream a module owns.
type StreamDeclaration struct {
go/membership/grants.go
+7
View File
@@ -12,6 +12,13 @@ import (
// cannot drift: until 2026-10-30 the grants part wrote its own wrapper ({node, module, digest, membership, published,
// retired}) there while the tool runner read the value as a bare membership, and served nothing (issue 774).
// GrantsBucket is the grants part's state memberships (buscompose.GrantsMembershipsBucket, which a test keeps agreeing),
// and GrantsKey an assignment's key in it, `<node>.<module>`: where a runtime reads its membership (FollowMembership).
const GrantsBucket = "mesh-grants_memberships"
// GrantsKey is an assignment's key in GrantsBucket.
func GrantsKey(node, module string) string { return node + "." + module }
// EncodeGrantsMembership is the value of a membership's key in the grants part's memberships: the bare membership.
func EncodeGrantsMembership(m Membership) []byte {
raw, _ := json.Marshal(m) // a Membership holds nothing json cannot write
go/process/follow.go
+34 -16
View File
@@ -1,24 +1,26 @@
package process
import (
"encoding/json"
"context"
"fmt"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"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
// each later one as it is issued, 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.
// 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.
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")
@@ -29,8 +31,8 @@ func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(er
}
}
take := func(body []byte) {
var issued membership.Membership
if err := json.Unmarshal(body, &issued); err != nil {
issued, err := membership.DecodeGrantsMembership(body)
if err != nil {
bad(fmt.Errorf("a membership arrived that is not one: %w", err))
return
}
@@ -40,14 +42,30 @@ func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(er
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)
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
}
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
}
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
}
go func() {
for e := range w.Updates() {
if e != nil && e.Operation() == jetstream.KeyValuePut {
take(e.Value())
}
}
}()
return nil
}