mesh-grants publishes memberships on change; the user list, assertions and grants stay in shadow (issue 1945)
Since release 6 nobody publishes memberships, so a module placed since cannot serve its seat or reach a new state (the operator, 2026-10-25 22:31). The part publishes each membership on the assignments stream only when it differs from the one last published, composes from the manifest the wanted build names, and says a refused publish and tries again at the next event. Pins mesh-sdk 881da7f, whose composer grants mesh-grants that publish.
This commit is contained in:
@@ -1,14 +1,21 @@
|
||||
// Command mesh-grants is the grants part (novox/hq ADR 0547 clauses 5 to 8, issue 1945), in shadow for the first
|
||||
// milestone: a process on its own bus credential that watches placement, the indexer's manifests, mesh-config's
|
||||
// resolved values and the node agents' haves, and on every change composes the bus's user list, each assignment's
|
||||
// membership and each manifest's buckets, logs and streams with the bus composer the controller uses (mesh-sdk
|
||||
// buscompose), compares them with what the bus loaded and holds (through the bus module's own read tools), and writes
|
||||
// only its own state. It publishes no membership, asserts nothing, writes no grant and no user list the bus node reads.
|
||||
// It holds no timer, and never exits for a refused read or write: a refusal is said and tried again at the next event.
|
||||
// Command mesh-grants is the grants part (novox/hq ADR 0547 clauses 5 to 8, issue 1945): a process on its own bus
|
||||
// credential that watches placement, the assigner's wants, the indexer's manifests, the builder's builds, mesh-config's
|
||||
// resolved values and the node agents' haves, and on every change composes each assignment's membership, the bus's user
|
||||
// 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. **The user list, the assertions and the grants stay in shadow** until
|
||||
// the operator's take-over, which this milestone always refuses: they are composed, compared with what the bus loaded
|
||||
// and holds (through the bus module's own read tools) and kept in the part's own state, and take no effect.
|
||||
//
|
||||
// It holds no timer, and never exits for a refused read, write or publish: a refusal is said and tried again at the next
|
||||
// event.
|
||||
//
|
||||
// **No seat in this milestone.** ADR 0547 clause 5 gives the part the seat `mesh-grants`; a `mesh-*` seat must be in
|
||||
// mesh-sdk's own seat list, which the running controller's composer does not know, so the part serves module tools
|
||||
// (grants_*) and claims no seat until take-over.
|
||||
// (grants_*) and claims no seat until take-over. The composer grants the membership publish by the module's name,
|
||||
// mesh-grants.
|
||||
//
|
||||
// Started by the tool runner, with no credential of its own, it serves those tools (a process never serves, ADR 0436).
|
||||
package main
|
||||
@@ -27,6 +34,7 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
@@ -103,9 +111,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return err
|
||||
}
|
||||
bus := newBusReader(ctx, js)
|
||||
f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn}}
|
||||
f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn},
|
||||
Publish: memberships{ctx: ctx, nc: nc, js: js}}
|
||||
f.Watch = bus
|
||||
logger.Printf("started on the account %s, in shadow: composing and comparing, writing only its own state", account.User)
|
||||
f.Asked = func(wake func()) error {
|
||||
return conn.State(grants.StateUserList).Watch(grants.KeyAsked, func(c stdio.StateChange) error {
|
||||
if !c.Current {
|
||||
wake()
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
logger.Printf("started on the account %s: publishing memberships; the user list, assertions and grants in shadow", account.User)
|
||||
f.Run(ctx)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
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.
|
||||
type memberships struct {
|
||||
ctx context.Context
|
||||
nc *nats.Conn
|
||||
js jetstream.JetStream
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -66,8 +66,8 @@ func toolsOf(p *grants.Part) []stdio.Tool {
|
||||
Input: schema(map[string]string{"module": "one module"}),
|
||||
Run: func(a map[string]any) (any, error) { return p.Grants(grants.Answer(a, "module")) }},
|
||||
{Name: grants.ToolMembership, Description: "One module's membership on one node (ADR 0547 clause 6): as composed now from the " +
|
||||
"bus's records with the bus composer, as last kept in the part's state, and as published (not readable from here: the bus " +
|
||||
"grants each module its own alone), with what the composition cannot read. Read-only.",
|
||||
"bus's records with the bus composer, and what the part's process published of it at its last pass (published, " +
|
||||
"unchanged, or why not), with what the composition cannot read. Read-only.",
|
||||
Input: schema(map[string]string{"node": "the machine", "module": "the module"}, "node", "module"),
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return p.MembershipOf(grants.Answer(a, "node"), grants.Answer(a, "module"))
|
||||
@@ -77,11 +77,13 @@ func toolsOf(p *grants.Part) []stdio.Tool {
|
||||
"they differ: users added, users dropped (and of those, gone from placement), each user's grant differences, the " +
|
||||
"assignments left out, and every input the bus does not carry, named. Read-only.",
|
||||
Run: func(map[string]any) (any, error) { return p.UserList() }},
|
||||
{Name: grants.ToolComposeNow, Description: "Compose again now (ADR 0547 clause 6), as the part does on every change it hears: the " +
|
||||
"user list, the memberships and the assertions, compared with what the bus loaded and holds; writes its own state only " +
|
||||
"where a value changed (one revision when an input changed, none when not) and says user-list-changed and " +
|
||||
"membership-changed for each. In shadow it publishes, asserts and grants nothing.",
|
||||
Run: func(map[string]any) (any, error) { return p.Compose(), nil }},
|
||||
{Name: grants.ToolComposeNow, Description: "Compose again now (ADR 0547 clause 6), as the part does on every change it hears: asks " +
|
||||
"the part's process (which alone holds the grant to publish a membership) to compose the memberships, the user list and " +
|
||||
"the assertions, publish each membership that differs from the one last published, and keep what changed in its own " +
|
||||
"state (one revision when an input changed, none when not), saying user-list-changed and membership-changed. Answers " +
|
||||
"the ask and the last pass; grants_user_list shows the pass that answers it. The user list, assertions and grants stay " +
|
||||
"in shadow.",
|
||||
Run: func(a map[string]any) (any, error) { return p.AskCompose(stdio.CallerOf(a)) }},
|
||||
{Name: grants.ToolAssert, Description: "One module's buckets, logs and streams (ADR 0547 clause 6), composed from its manifest at " +
|
||||
"the indexer's trunk with their limits, beside what the bus holds (through the bus module's nats_streams). In shadow it " +
|
||||
"asserts nothing, and says so.",
|
||||
|
||||
Reference in New Issue
Block a user