diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 378c10b..9412fac 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -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) diff --git a/go/buscompose/creates_streams_test.go b/go/buscompose/creates_streams_test.go index 73047bf..efc3efb 100644 --- a/go/buscompose/creates_streams_test.go +++ b/go/buscompose/creates_streams_test.go @@ -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. diff --git a/go/buscompose/nats.go b/go/buscompose/nats.go index 85dbab2..86aa45e 100644 --- a/go/buscompose/nats.go +++ b/go/buscompose/nats.go @@ -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. diff --git a/go/buscompose/streams.go b/go/buscompose/streams.go index fd2dd42..6af9e61 100644 --- a/go/buscompose/streams.go +++ b/go/buscompose/streams.go @@ -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 diff --git a/go/catalogue/streams.go b/go/catalogue/streams.go index fa166e2..40a4ff4 100644 --- a/go/catalogue/streams.go +++ b/go/catalogue/streams.go @@ -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 { diff --git a/go/membership/grants.go b/go/membership/grants.go index b6d4a30..b14f66d 100644 --- a/go/membership/grants.go +++ b/go/membership/grants.go @@ -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, `.`: 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 diff --git a/go/process/follow.go b/go/process/follow.go index 07678bc..9c1fc33 100644 --- a/go/process/follow.go +++ b/go/process/follow.go @@ -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 }