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

Merge pull request 'mesh-grants: create a declared module's missing buckets, logs and streams on the event; publish memberships past retired seat verbs (issues 1790, 1945)' (#58) from fix/1790-grants-create-buckets into main

Browse Source
This commit was merged in pull request #58.
This commit is contained in:
mesh-tools
2026-10-26 08:17:22 +00:00
parent 13380e3d0e 12f16bbebd
commit 86db64904b
13 changed files with 579 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/cmd/mesh-grants/create.go
+34
View File
@@ -0,0 +1,34 @@
package main
import (
"context"
"errors"
"fmt"
"github.com/nats-io/nats.go/jetstream"
"mesh-grants/internal/grants"
)
// creating creates a stream a manifest declares that the bus lacks (issue 1790; the operator, 2026-10-26 08:54): the
// create alone, which the account is granted by the stream's name, never an update or a delete. The one place the part
// creates on the bus.
type creating struct {
ctx context.Context
js jetstream.JetStream
}
func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error {
ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
defer cancel()
_, err := c.js.CreateStream(ctx, cfg)
switch {
case err == nil:
return nil
case errors.Is(err, jetstream.ErrStreamNameAlreadyInUse):
return grants.ErrHeld
case errors.Is(err, context.DeadlineExceeded):
return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err)
}
return err
}
modules/mesh-grants/cmd/mesh-grants/main.go
+6 -3
View File
@@ -5,7 +5,10 @@
//
// **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
// 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, the rest of 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.
//
@@ -112,7 +115,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}}
Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js}}
f.Watch = bus
f.Asked = func(wake func()) error {
return conn.State(grants.StateUserList).Watch(grants.KeyAsked, func(c stdio.StateChange) error {
@@ -122,7 +125,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return nil
})
}
logger.Printf("started on the account %s: publishing memberships; the user list, assertions and grants in shadow", account.User)
logger.Printf("started on the account %s: publishing memberships and creating the declared buckets, logs and streams the bus lacks; the user list, grants and any change to a held stream in shadow", account.User)
f.Run(ctx)
return nil
}
modules/mesh-grants/cmd/mesh-grants/tools.go
+8 -5
View File
@@ -67,9 +67,10 @@ func toolsOf(p *grants.Part) []stdio.Tool {
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, and what the part's process published of it at its last pass (published, " +
"unchanged, or why not: a shadow difference names what publishing would lose), with what the composition cannot " +
"read. Read-only.",
Input: schema(map[string]string{"node": "the machine", "module": "the module"}, "node", "module"),
"unchanged, or why not: a shadow difference names what publishing would lose; verbs the seat catalogue retired are " +
"dropped and named), with what the composition cannot read. With neither node nor module: every membership the last " +
"pass held back, each with what it would lose. Read-only.",
Input: schema(map[string]string{"node": "the machine (with module)", "module": "the module (with node)"}),
Run: func(a map[string]any) (any, error) {
return p.MembershipOf(grants.Answer(a, "node"), grants.Answer(a, "module"))
}},
@@ -90,8 +91,10 @@ func toolsOf(p *grants.Part) []stdio.Tool {
"verb, and the answer says so.",
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.",
"the indexer's trunk with their limits, beside what the bus holds (through the bus module's nats_streams). When the " +
"bus lacks one, asks the part's process (which alone holds the grant) to create it now with the manifest's limits, " +
"as it does by itself when a manifest is registered (issue 1790), and says asserted; one the bus holds is never " +
"changed, a difference is only said. Answers the comparison and the ask; call again to see what the bus holds.",
Input: schema(map[string]string{"module": "the module"}, "module"),
Run: func(a map[string]any) (any, error) { return p.AssertOne(grants.Answer(a, "module")) }},
{Name: grants.ToolTakeOver, Description: "The operator's switch to the grants part's list (ADR 0547 clause 8), refused without the " +
modules/mesh-grants/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-grants
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025215021-b7b5c6ef27a0
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026080443-136fa8b2a896
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-grants/go.sum
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025215021-b7b5c6ef27a0 h1:8dKGrZzPthx/jZlV9Bjm2lWGNORUTETmq2vxKN+MRV0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025215021-b7b5c6ef27a0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026080443-136fa8b2a896 h1:5rHfru5NJ7SK20QOt8YUJSglcn1gsjtLc9AGmWYGnRQ=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026080443-136fa8b2a896/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-grants/grants_test.go
+193 -7
View File
@@ -14,6 +14,8 @@ import (
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -43,6 +45,9 @@ type fakeBus struct {
refusePub bool
// hold, when set, keeps every nats_users call waiting until it is closed: a pass waiting on the bus.
hold chan struct{}
// creates are the streams the part asked to create, refused when refuseCreate.
creates []string
refuseCreate bool
}
func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) {
@@ -72,6 +77,25 @@ func (b *fakeBus) PublishMembership(node, module string, body []byte, after uint
return nil
}
// CreateDeclared is the bus's create of a stream: refused when refuseCreate, held when the bus holds it, and otherwise
// held from now on, as nats_streams then reads it.
func (b *fakeBus) CreateDeclared(cfg jetstream.StreamConfig) error {
b.mu.Lock()
defer b.mu.Unlock()
b.creates = append(b.creates, cfg.Name)
if b.refuseCreate {
return errors.New("Permissions Violation for Publish to $JS.API.STREAM.CREATE." + cfg.Name)
}
for _, s := range b.streams {
if s["name"] == cfg.Name {
return grants.ErrHeld
}
}
b.streams = append(b.streams, map[string]any{"name": cfg.Name, "subjects": cfg.Subjects, "retention": "limits",
"max_age_hours": int(cfg.MaxAge.Hours()), "max_per_subject": int(cfg.MaxMsgsPerSubject)})
return nil
}
func newFakeBus() *fakeBus {
return &fakeBus{states: map[string]map[string][]byte{}, refused: map[string]bool{}, own: map[string]map[string][]byte{},
revs: map[string]uint64{}, refuseTo: map[string]bool{}}
@@ -295,12 +319,14 @@ func TestGrants(t *testing.T) {
}
})
t.Run("no timer, no create, a publish only of memberships", func(t *testing.T) {
t.Run("no timer, no update or delete, a publish only of memberships, a create only of declared streams", func(t *testing.T) {
forbidden := []string{"time.Tick", "time.NewTicker", "time.Sleep", "time.After", "time.NewTimer", "time.AfterFunc",
"PublishMsg", "CreateStream", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue",
"CreateObjectStore", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
// The one publish: a membership, on its own subject, in publish.go.
"PublishMsg", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue", "DeleteStream", "PurgeStream",
"DeleteKeyValue", "CreateObjectStore", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
// The one publish: a membership, on its own subject, in publish.go; the one create: a declared stream the bus lacks,
// in create.go (issue 1790).
publishOnly := filepath.Join("cmd", "mesh-grants", "publish.go")
createOnly := filepath.Join("cmd", "mesh-grants", "create.go")
files := 0
for _, root := range []string{"cmd", "internal"} {
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
@@ -315,6 +341,9 @@ func TestGrants(t *testing.T) {
if path == publishOnly && strings.Count(string(src), ".Publish(") != 1 {
t.Errorf("publish.go publishes other than one membership")
}
if n := strings.Count(string(src), "CreateStream"); (path == createOnly && n != 1) || (path != createOnly && n != 0) {
t.Errorf("%s holds CreateStream %d times: only create.go creates, once", path, n)
}
for _, f := range forbidden {
if strings.Contains(string(src), f) {
t.Errorf("%s holds %s: the part holds no timer and creates or asserts nothing", path, f)
@@ -513,8 +542,9 @@ func TestGrants(t *testing.T) {
}
a, err := p.AssertOne("mod-x")
as, _ := a.(grants.Assertion)
if err != nil || as.Asserted || as.Says != grants.ShadowAsserts || len(as.Items) != 3 {
am, _ := a.(map[string]any)
as, _ := am["assertion"].(grants.Assertion)
if err != nil || as.Asserted || as.Says != grants.ShadowAsserts || len(as.Items) != 3 || am["asked"] == nil {
t.Fatalf("assert = %+v, %v", a, err)
}
for _, it := range as.Items {
@@ -639,6 +669,17 @@ func TestGrants(t *testing.T) {
m["serves"] = append(m["serves"].([]any), map[string]any{"subject": "mesh.mod.mod-x.tool.gone"})
return m
}, false, "tool.gone"},
// A verb the seat catalogue retired on purpose (mark-ready, mesh-sdk d506f21, ADR 0421) is dropped and named; a
// verb of a seat the catalogue does not define is still a loss (issue 1945).
{"a change that drops only a retired seat verb is published, the verb named", func(m map[string]any) map[string]any {
m["seats"] = []any{map[string]any{"seat": "mesh-delivery", "verb": "mark-ready", "subject": "mesh.seat.mesh-delivery.tool.mark-ready"}}
return m
}, true, "mesh-delivery mark-ready"},
{"a retired verb and a module seat's verb dropped: not published, the module seat's named", func(m map[string]any) map[string]any {
m["seats"] = []any{map[string]any{"seat": "mesh-delivery", "verb": "mark-ready", "subject": "mesh.seat.mesh-delivery.tool.mark-ready"},
map[string]any{"seat": "telegram-sender", "verb": "send", "subject": "mesh.seat.telegram-sender.tool.send"}}
return m
}, false, "telegram-sender"},
} {
t.Run(row.name, func(t *testing.T) {
b := fixture()
@@ -653,9 +694,12 @@ func TestGrants(t *testing.T) {
t.Fatalf("published %v, not published %v", res.Published, res.NotPublished)
}
if row.published {
if row.last != nil && !strings.Contains(string(b.stream["mesh.assignment."+key]), "10.0.0.1") {
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 raw, _, _, _ := b.ownGet(grants.StateMemberships, key); !strings.Contains(string(raw), row.names) {
t.Errorf("the pass does not name %q: %s", row.names, raw)
}
return
}
said := strings.Join(res.NotPublished, "\n")
@@ -707,6 +751,138 @@ func TestGrants(t *testing.T) {
})
})
t.Run("assertions: what the bus lacks is created once, what it holds is never touched", func(t *testing.T) {
b := fixture()
b.loaded = shadowOf(t, b)
// The bus holds mod-x's bucket with another history, and neither its log nor its stream.
b.streams = []map[string]any{{"name": "KV_mod-x_kept", "subjects": []string{"$KV.mod-x_kept.>"}, "retention": "limits", "max_per_subject": 2}}
p := partOn(b)
p.Create = b
assertion := func() grants.Assertion {
raw, _, _, _ := b.ownGet(grants.StateAssertions, "mod-x")
var a grants.Assertion
_ = json.Unmarshal(raw, &a)
return a
}
first := p.Compose()
if !slices.Equal(b.creates, []string{"LOG_mod-x_history", "THINGS"}) || len(first.Asserted) != 1 ||
!strings.HasPrefix(first.Asserted[0], "mod-x at ") || !slices.Contains(b.events, grants.EventAsserted) {
t.Fatalf("the first pass created %v, said %v, events %v", b.creates, first.Asserted, b.events)
}
a := assertion()
if !a.Asserted || a.Says != grants.CreatesMissing {
t.Errorf("the assertion kept = %+v", a)
}
for _, it := range a.Items {
switch it.Stream {
case "KV_mod-x_kept":
if it.Created != "" || !slices.Contains(it.Differences, "history 2 held, 3 composed") {
t.Errorf("a bucket the bus holds was touched, or its difference not said: %+v", it)
}
default:
if it.Created != grants.ItemCreated {
t.Errorf("%s: %q, want created", it.Stream, it.Created)
}
}
}
if again := p.Compose(); len(again.Asserted) != 0 || len(b.creates) != 2 {
t.Errorf("a pass with everything held created again: %v, %v", again.Asserted, b.creates)
}
// THINGS deleted since: not made again at the same manifest, said; a hand assert (the tool, asking the process)
// makes it again.
b.mu.Lock()
b.streams = slices.DeleteFunc(b.streams, func(s map[string]any) bool { return s["name"] == "THINGS" })
b.mu.Unlock()
gone := p.Compose()
if len(b.creates) != 2 || !strings.Contains(strings.Join(gone.NotAsserted, " "), "THINGS: created at this manifest and gone since") {
t.Errorf("a stream gone since was made again by itself, or not said: %v, %v", b.creates, gone.NotAsserted)
}
tool := partOn(b)
if got, err := tool.AssertOne("mod-x"); err != nil || fmt.Sprint(got) == "" || len(b.creates) != 2 {
t.Fatalf("the tool = %v, %v, creates %v: it asks, and creates nothing itself", got, err, b.creates)
}
if hand := p.Compose(); len(hand.Asserted) != 1 || b.creates[len(b.creates)-1] != "THINGS" {
t.Errorf("the hand assert did not create THINGS again: %v, %v", hand.Asserted, b.creates)
}
if after := p.Compose(); len(after.Asserted) != 0 {
t.Errorf("the hand ask was acted on twice: %v", after.Asserted)
}
for _, c := range b.creates {
if c == "KV_mod-x_kept" {
t.Error("the bucket the bus holds was asked to be created")
}
}
})
t.Run("assertions: a refused create is said, never stops it, and is tried at the next event", func(t *testing.T) {
b := fixture()
b.loaded = shadowOf(t, b)
b.refuseCreate = true
p := partOn(b)
p.Create = b
refused := p.Compose()
if len(refused.Asserted) != 0 || !strings.Contains(strings.Join(refused.NotAsserted, " "), "Permissions Violation") ||
slices.Contains(b.events, grants.EventAsserted) {
t.Fatalf("a refused create = %+v", refused)
}
b.refuseCreate = false
if next := p.Compose(); len(next.Asserted) != 1 || !strings.Contains(next.Asserted[0], "KV_mod-x_kept") {
t.Errorf("the next event created %v", next.Asserted)
}
// What the bus holds unread: nothing is created.
c := fixture()
c.loaded = shadowOf(t, c)
q := partOn(c)
q.Create = c
q.Call = refusingStreams{c}
if r := q.Compose(); len(c.creates) != 0 || len(r.Asserted) != 0 {
t.Errorf("created %v with what the bus holds unread", c.creates)
}
})
t.Run("memberships held back: every one is said and listed", func(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"} {
raw, _ := json.Marshal(grants.Memberships(recs)[key].Membership)
var m map[string]any
_ = json.Unmarshal(raw, &m)
m["serves"] = append(m["serves"].([]any), map[string]any{"subject": "mesh.mod.lost.tool.gone"})
raw, _ = json.Marshal(m)
subject := "mesh.assignment." + key
b.stream[subject], b.seqs[subject] = raw, uint64(i+1)
}
var logged []string
var lmu sync.Mutex
w := &fakeWatcher{opened: make(chan string, 16)}
f := &grants.Follower{Part: partOn(b), Watch: w, Logf: func(format string, args ...any) {
lmu.Lock()
logged = append(logged, fmt.Sprintf(format, args...))
lmu.Unlock()
}}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { f.Run(ctx); close(done) }()
waitFor(t, func() bool {
lmu.Lock()
defer lmu.Unlock()
all := strings.Join(logged, "\n")
return strings.Contains(all, "held a.mod-x: ") && strings.Contains(all, "held b.mod-y: ")
})
cancel()
<-done
held, err := partOn(b).MembershipOf("", "")
h, _ := held.(map[string]any)
if err != nil || h["count"] != 2 || !strings.Contains(fmt.Sprint(h["held"]), "tool.gone") {
t.Errorf("grants_membership with neither node nor module = %v, %v", held, err)
}
if _, err := partOn(b).MembershipOf("a", ""); err == nil {
t.Error("a node without a module was answered")
}
})
t.Run("a refused read or write never stops it", func(t *testing.T) {
b := fixture()
b.loaded = shadowOf(t, b)
@@ -756,6 +932,16 @@ func nonNil(s []string) []string {
}
// fakeWatcher refuses a state's first watches as told, and hands changes to the watches it opened.
// refusingStreams is the bus with nats_streams refused: what the bus holds cannot be read.
type refusingStreams struct{ *fakeBus }
func (r refusingStreams) Call(key string, body any) (json.RawMessage, error) {
if key == grants.CallStreams {
return nil, errors.New("refused")
}
return r.fakeBus.Call(key, body)
}
type fakeWatcher struct {
mu sync.Mutex
refuse map[string]int
modules/mesh-grants/internal/grants/assertions.go
+121 -3
View File
@@ -2,9 +2,14 @@ package grants
import (
"encoding/json"
"errors"
"fmt"
"slices"
"sort"
"strings"
"sync"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -19,6 +24,9 @@ type Item struct {
Held *Limits `json:"held,omitempty"`
// Differences says how they differ, in words; empty when they agree in what can be read from here.
Differences []string `json:"differences,omitempty"`
// Created says what became of creating it when the bus lacked it: "created", or why not (issue 1790). Empty when
// the bus holds it or no creation was asked here.
Created string `json:"created,omitempty"`
}
// Limits are the facts of a stream an assertion sets and the bus module's nats_streams reads back. Its size cap is not
@@ -31,8 +39,8 @@ type Limits struct {
MaxBytes int64 `json:"max_bytes,omitempty"`
}
// Assertion is one module's declared buckets, logs and streams, composed and compared. Asserted is always false in
// shadow: nothing is created or changed.
// Assertion is one module's declared buckets, logs and streams, composed and compared. Asserted says this pass created
// one the bus lacked (the operator, 2026-10-26 08:54); one the bus holds is never changed.
type Assertion struct {
Module string `json:"module"`
Manifest string `json:"manifest"`
@@ -41,9 +49,119 @@ type Assertion struct {
Says string `json:"says"`
}
// ShadowAsserts is what every assertion says in this milestone.
// ShadowAsserts is what an assertion says where nothing may be created: in the tools, which ask the part's process.
const ShadowAsserts = "in shadow: composed and compared, nothing asserted (ADR 0547 clause 5 asserts from take-over)"
// CreatesMissing is what an assertion of the part's process says (the operator, 2026-10-26 08:54; issue 1790).
const CreatesMissing = "a bucket, log or stream the manifest declares and the bus lacks is created with the manifest's limits, once per " +
"manifest; one the bus holds is never changed or deleted, a difference is only said (ADR 0547 clause 5 asserts the rest from take-over)"
// Creator creates one stream the bus does not hold, with the configuration given, and never updates or deletes one: the
// part's process alone, whose account is granted the create of each declared stream by name (the operator, 2026-10-26
// 08:54). A stream the bus holds already answers ErrHeld.
type Creator interface {
CreateDeclared(cfg jetstream.StreamConfig) error
}
// ErrHeld is a create refused because the bus holds the stream already (made meanwhile): left as it is.
var ErrHeld = errors.New("the bus holds it already, so it is left as it is")
// Created words of an item.
const (
ItemCreated = "created"
ItemNotCreated = "not created: "
)
// Configs are the streams a manifest declares, as the controller asserts them at its start (mesh-sdk buscompose), by
// stream name. Pure.
func Configs(m catalogue.Manifest) map[string]jetstream.StreamConfig {
out := map[string]jetstream.StreamConfig{}
for _, b := range bucketsOf(m) {
c := buscompose.BucketStreamConfig(b)
out[c.Name] = c
}
for _, l := range logsOf(m) {
c := l.Config()
out[c.Name] = c
}
for _, s := range streamsOf(m) {
c := buscompose.ModuleStreamConfig(s)
out[c.Name] = c
}
return out
}
// creations remembers each stream created, by the manifest digest it was created at: created once per manifest, so a
// stream deleted since is not made again until a new manifest or a hand assert (grants_assert).
type creations struct {
mu sync.Mutex
made map[string]string
}
func (c *creations) madeAt(stream string) string {
c.mu.Lock()
defer c.mu.Unlock()
return c.made[stream]
}
func (c *creations) keep(stream, digest string) {
c.mu.Lock()
defer c.mu.Unlock()
if c.made == nil {
c.made = map[string]string{}
}
c.made[stream] = digest
}
// createMissing creates each item of a the bus lacks, with the manifest's limits, and says what became of it. It never
// touches an item the bus holds, and does nothing when what the bus holds was not read. force asks again a stream
// created at this manifest and gone since (a hand assert). It answers the streams created.
func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifest, force bool) []string {
cfgs := Configs(m)
var made []string
a.Says = CreatesMissing
for i := range a.Items {
it := &a.Items[i]
if it.Held != nil || !slices.Contains(it.Differences, "the bus holds no stream "+it.Stream) {
continue
}
if !force && done.madeAt(it.Stream) == a.Manifest {
it.Created = ItemNotCreated + "created at this manifest and gone since; grants_assert creates it again"
continue
}
cfg, ok := cfgs[it.Stream]
if !ok {
it.Created = ItemNotCreated + "no configuration composed for it"
continue
}
switch err := c.CreateDeclared(cfg); {
case err == nil:
it.Created = ItemCreated
done.keep(it.Stream, a.Manifest)
made = append(made, it.Kind+" "+it.Stream)
case errors.Is(err, ErrHeld):
it.Created = ItemNotCreated + ErrHeld.Error()
default:
it.Created = ItemNotCreated + err.Error() + " (tried again at the next event)"
}
}
if len(made) > 0 {
a.Asserted = true
}
return made
}
// notCreated is each item of a creation refused, with why, as `<stream>: <why>`.
func notCreated(a Assertion) []string {
var out []string
for _, it := range a.Items {
if why, ok := strings.CutPrefix(it.Created, ItemNotCreated); ok && it.Created != "" {
out = append(out, it.Stream+": "+why)
}
}
return out
}
// Composed is a module's buckets, logs and streams as an assertion would ask for them. Pure.
func Composed(m catalogue.Manifest) []Item {
var out []Item
modules/mesh-grants/internal/grants/follow.go
+32 -3
View File
@@ -3,6 +3,7 @@ package grants
import (
"context"
"encoding/json"
"strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -35,6 +36,9 @@ type Follower struct {
watched map[string]bool
refused map[string]string
have map[string]string
// held and notAsserted are what the last pass held back and could not create, said again only when they change.
held string
notAsserted string
}
func (f *Follower) init() {
@@ -142,6 +146,33 @@ func (f *Follower) open() {
}
}
// said logs every membership the pass held back, each with what it would lose (issue 1945), and every declared stream
// the pass could not create, each with why (issue 1790), when the set changes from the last pass's; and every stream it
// created, one line per module.
func (f *Follower) said(res Result) {
if held := strings.Join(res.NotPublished, "\n"); held != f.held {
f.held = held
if len(res.NotPublished) == 0 {
f.logf("no membership held back")
} else {
f.logf("%d memberships not published, tried again at the next event (grants_membership with no node and module lists them):",
len(res.NotPublished))
for _, n := range res.NotPublished {
f.logf("held %s", n)
}
}
}
for _, a := range res.Asserted {
f.logf("asserted %s", a)
}
if not := strings.Join(res.NotAsserted, "\n"); not != f.notAsserted {
f.notAsserted = not
for _, n := range res.NotAsserted {
f.logf("not asserted %s", n)
}
}
}
// Run composes at start and after every change until ctx ends. It returns only when ctx ends.
func (f *Follower) Run(ctx context.Context) {
f.init()
@@ -159,9 +190,7 @@ func (f *Follower) Run(ctx context.Context) {
if len(res.Published) > 0 {
f.logf("published %d memberships: %v", len(res.Published), res.Published)
}
if len(res.NotPublished) > 0 {
f.logf("%d memberships not published, tried again at the next event; the first: %s", len(res.NotPublished), res.NotPublished[0])
}
f.said(res)
if res.Written {
f.logf("the shadow list moved to revision %d: %d users, %d differing from the loaded list",
res.Revision, res.UserList.Users, len(res.UserList.Compare.Differing()))
modules/mesh-grants/internal/grants/inputs.go
+5 -3
View File
@@ -1,8 +1,10 @@
// Package grants is the grants part in shadow (novox/hq ADR 0547 clauses 5 to 8, issue 1945): it composes the bus's user
// list, each module's membership and each manifest's buckets, logs and streams from what the bus holds, with the bus
// composer the controller uses (mesh-sdk buscompose), and compares them with what the bus loaded and holds. It writes only
// its own state and takes no effect: it publishes no membership, asserts nothing, writes no grant and no user list the
// bus node reads, until the operator's take-over, which this milestone always refuses.
// composer the controller uses (mesh-sdk buscompose), and compares them with what the bus loaded and holds. Beyond its own
// state it takes effect in two things alone, each by a grant the operator allowed: it publishes memberships (2026-10-25
// 22:31), and creates a bucket, log or stream a manifest declares that the bus lacks (2026-10-26 08:54, issue 1790). It
// changes no stream the bus holds, writes no grant and no user list the bus node reads, until the operator's take-over,
// which this milestone always refuses.
package grants
import (
modules/mesh-grants/internal/grants/memberships.go
+62 -1
View File
@@ -11,6 +11,7 @@ import (
"strings"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/membership"
)
@@ -41,6 +42,9 @@ type Membership struct {
// Published says what became of it at the last pass: "published", "unchanged" (equal to the one last published),
// or why it was not.
Published string `json:"published"`
// Retired are the seat verbs the last membership served that mesh-sdk's seat catalogue no longer has, `<seat>
// <verb>`: dropped on purpose, so their loss does not hold the membership back (issue 1945).
Retired []string `json:"retired,omitempty"`
}
// Memberships composes every assignment's membership from the records (buscompose.MembershipFor, as the controller's
@@ -101,7 +105,9 @@ func publish(p Publisher, m Membership) (Membership, bool) {
// composed with no root-free judgement, a kind's verified-sender would be dropped, and the operator's approvals by
// a channel would no longer be honoured. Any loss stays a shadow difference, named.
if found {
if lost := Losses(was, body); len(lost) > 0 {
kept, retired := WithoutRetired(was)
m.Retired = retired
if lost := Losses(kept, body); len(lost) > 0 {
m.Published = ShadowDifference + strings.Join(lost, "; ")
return m, false
}
@@ -114,6 +120,61 @@ func publish(p Publisher, m Membership) (Membership, bool) {
return m, true
}
// WithoutRetired is a membership without the seat verbs it serves that mesh-sdk's seat catalogue no longer has (a verb
// retired on purpose, such as the delivery seat's mark-ready, ADR 0421), and those verbs, `<seat> <verb>`, once each. A
// seat the catalogue does not define (a module's own) keeps every verb, so its loss is still refused. Pure.
func WithoutRetired(last []byte) ([]byte, []string) {
var m map[string]any
if json.Unmarshal(last, &m) != nil {
return last, nil
}
seats, ok := m["seats"].([]any)
if !ok {
return last, nil
}
var kept []any
var retired []string
for _, e := range seats {
served, _ := e.(map[string]any)
seat, _ := served["seat"].(string)
verb, _ := served["verb"].(string)
if retiredVerb(seat, verb) {
if r := seat + " " + verb; !slices.Contains(retired, r) {
retired = append(retired, r)
}
continue
}
kept = append(kept, e)
}
if len(retired) == 0 {
return last, nil
}
if len(kept) == 0 {
delete(m, "seats") // as an empty list is written: not at all
} else {
m["seats"] = kept
}
out, err := json.Marshal(m)
if err != nil {
return last, nil
}
sort.Strings(retired)
return out, retired
}
// retiredVerb says the seat catalogue defines seat with its verbs and no longer has verb among them. A seat the catalogue
// defines without verbs (its holder declares them) retires none.
func retiredVerb(seat, verb string) bool {
if seat == "" || verb == "" {
return false
}
s, ok := catalogue.SeatNamed(seat)
if !ok || len(s.Serves) == 0 {
return false
}
return !slices.ContainsFunc(s.Serves, func(v catalogue.Verb) bool { return v.Name == verb })
}
// sameMembership says two memberships say the same, whatever the order of their fields. Pure.
func sameMembership(was, now []byte) bool {
var a, b any
modules/mesh-grants/internal/grants/part.go
+59 -14
View File
@@ -35,10 +35,12 @@ var States = []string{StateAssertions, StateMemberships, StateTakeover, StateUse
const (
EventUserList = "user-list-changed"
EventMembership = "membership-changed"
// EventAsserted is said when the part created a bucket, log or stream a manifest declares (issue 1790).
EventAsserted = "asserted"
)
// Emits is every event the part says.
var Emits = []string{EventMembership, EventUserList}
var Emits = []string{EventAsserted, EventMembership, EventUserList}
// Shadow is the switch's one state in this milestone: the bus node follows the controller's list.
const Shadow = "shadow"
@@ -83,21 +85,25 @@ func (s UserListState) inputs() string {
// Result is what one composition did.
type Result struct {
UserList UserListState `json:"user_list"`
Revision uint64 `json:"revision"`
Written bool `json:"written"`
Memberships int `json:"memberships"`
MembershipsWritten int `json:"memberships_written"`
Published []string `json:"published,omitempty"`
NotPublished []string `json:"not_published,omitempty"`
Assertions int `json:"assertions"`
AssertionsWritten int `json:"assertions_written"`
Errors []string `json:"errors,omitempty"`
PublishedOrAsserted string `json:"published_or_asserted"`
UserList UserListState `json:"user_list"`
Revision uint64 `json:"revision"`
Written bool `json:"written"`
Memberships int `json:"memberships"`
MembershipsWritten int `json:"memberships_written"`
Published []string `json:"published,omitempty"`
NotPublished []string `json:"not_published,omitempty"`
Assertions int `json:"assertions"`
AssertionsWritten int `json:"assertions_written"`
// Asserted is each module whose declared streams this pass created, `<module> at <digest>: created …`, and
// NotAsserted each creation refused, `<module> <stream>: <why>` (issue 1790).
Asserted []string `json:"asserted,omitempty"`
NotAsserted []string `json:"not_asserted,omitempty"`
Errors []string `json:"errors,omitempty"`
PublishedOrAsserted string `json:"published_or_asserted"`
}
// NothingTakesEffect is what every composition says it did on the bus beyond its own state and the memberships.
const NothingTakesEffect = "memberships only (on the assignments stream, a first one or one that only adds to the last published; a loss stays a shadow difference); the user list, assertions and grants stay in shadow until take-over"
const NothingTakesEffect = "memberships (on the assignments stream, a first one or one that only adds to the last published or drops only a verb the seat catalogue retired; any other loss 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 {
@@ -107,11 +113,36 @@ type Part struct {
Emit Emitter
// Publish publishes memberships: the process's connection; nil in the tools, which ask the process instead.
Publish Publisher
Now func() time.Time
// Create creates a declared stream the bus lacks: the process's connection; nil in the tools, which ask the process.
Create Creator
Now func() time.Time
cache Manifests
mu sync.Mutex
last Result
made creations
// askSeen is the revision of the last ask to compose (KeyAsked) this part acted on.
askSeen uint64
}
// Ask is the value of KeyAsked: a tool's ask to compose now, and, from grants_assert, to create one module's missing
// streams again even where they were created at this manifest before.
type Ask struct {
By string `json:"by"`
At time.Time `json:"at"`
Assert string `json:"assert,omitempty"`
}
// askedAssert is the module a new ask names to assert by hand, read once per ask; empty when none.
func (p *Part) askedAssert() string {
raw, rev, found, err := p.Store.Get(StateUserList, KeyAsked)
if err != nil || !found || rev == p.askSeen {
return ""
}
p.askSeen = rev
var a Ask
_ = json.Unmarshal(raw, &a)
return a.Assert
}
func (p *Part) now() time.Time {
@@ -187,11 +218,25 @@ func (p *Part) Compose() Result {
}
}
held, heldErr := HeldStreams(p.Call)
forced := ""
if p.Create != nil {
forced = p.askedAssert()
}
for _, module := range sortedKeys(in.Manifests) {
if !assigned(in, module) {
continue
}
a := Assert(module, in.Digests[module], in.Manifests[module], held, heldErr)
// Created only where the bus was read and lacks it, never changed where it holds it (issue 1790).
if p.Create != nil && heldErr == nil {
if made := createMissing(p.Create, &p.made, &a, in.Manifests[module], forced == module); len(made) > 0 {
res.Asserted = append(res.Asserted, fmt.Sprintf("%s at %s: created %s", module, a.Manifest, strings.Join(made, ", ")))
p.say(&res, EventAsserted, map[string]any{"module": module, "manifest": a.Manifest, "created": made})
}
for _, why := range notCreated(a) {
res.NotAsserted = append(res.NotAsserted, module+" "+why)
}
}
res.Assertions++
data, _ := json.Marshal(a)
if p.changed(&res, StateAssertions, module, data) {
modules/mesh-grants/internal/grants/tools.go
+55 -4
View File
@@ -62,8 +62,11 @@ func (p *Part) Grants(module string) (any, error) {
// MembershipOf answers the tool membership: one module on one node as composed now, and as last published by the
// part's process (its state memberships says what became of each at the last pass).
func (p *Part) MembershipOf(node, module string) (any, error) {
if node == "" && module == "" {
return p.Held()
}
if node == "" || module == "" {
return nil, errors.New("membership names a node and a module")
return nil, errors.New("membership names a node and a module, or neither for every membership held back")
}
in := ReadInputs(p.Read, &p.cache)
recs, left := Records(in)
@@ -90,6 +93,32 @@ func (p *Part) MembershipOf(node, module string) (any, error) {
return out, nil
}
// Held answers every membership the part's process held back at its last pass, each `<node>.<module>` with why (what
// publishing would lose), from the pass it keeps (issue 1945).
func (p *Part) Held() (any, error) {
raw, rev, found, err := p.Store.Get(StateUserList, KeyPass)
if err != nil {
return nil, fmt.Errorf("the last pass cannot be read: %w", err)
}
if !found {
return map[string]any{"held": []any{}, "says": "no pass of the part's process has been kept yet"}, nil
}
var res Result
if err := json.Unmarshal(raw, &res); err != nil {
return nil, fmt.Errorf("the last pass cannot be read: %w", err)
}
type held struct {
Membership string `json:"membership"`
Why string `json:"why"`
}
out := []held{}
for _, n := range res.NotPublished {
key, why, _ := strings.Cut(n, ": ")
out = append(out, held{Membership: key, Why: why})
}
return map[string]any{"held": out, "count": len(out), "pass": rev, "published": res.Published}, nil
}
// Pass composes and keeps the result as the last pass, for the tools to read.
func (p *Part) Pass() Result {
res := p.Compose()
@@ -108,7 +137,7 @@ const NoGrantAgain = "missing: a verb to ask the controller to write a module's
// AskCompose is the tool compose-now: it asks the part's process to compose and publish now (only the process holds the
// grant to publish a membership), and answers the last pass, whose revision moves when the process answered.
func (p *Part) AskCompose(caller string) (any, error) {
ask, _ := json.Marshal(map[string]any{"by": caller, "at": p.now()})
ask, _ := json.Marshal(Ask{By: caller, At: p.now()})
rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
if err != nil {
return nil, fmt.Errorf("the ask could not be written: %w", err)
@@ -151,7 +180,9 @@ func (p *Part) UserList() (any, error) {
}
// AssertOne answers the tool assert: one module's buckets, logs and streams as composed from its manifest at the
// indexer's trunk, beside what the bus holds. It asserts nothing.
// indexer's trunk, beside what the bus holds, and asks the part's process (which alone holds the grant to create) to
// create what the bus lacks now, even one it created at this manifest and that is gone since (issue 1790). It changes no
// stream the bus holds. Here, in the tools, it creates nothing itself: the answer names the ask.
func (p *Part) AssertOne(module string) (any, error) {
if module == "" {
return nil, errors.New("assert names a module")
@@ -172,7 +203,27 @@ func (p *Part) AssertOne(module string) (any, error) {
return nil, err
}
held, heldErr := HeldStreams(p.Call)
return Assert(module, v.Manifest, m, held, heldErr), nil
a := Assert(module, v.Manifest, m, held, heldErr)
out := map[string]any{"assertion": a}
missing := 0
for _, it := range a.Items {
if it.Held == nil {
missing++
}
}
if missing == 0 {
out["says"] = "the bus holds every bucket, log and stream the manifest declares: nothing to create; a difference is only said"
return out, nil
}
ask, _ := json.Marshal(Ask{By: "grants_assert", At: p.now(), Assert: module})
rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
if err != nil {
return nil, fmt.Errorf("the ask to create could not be written: %w", err)
}
out["asked"] = rev
out["says"] = fmt.Sprintf("%d the bus lacks: the part's process creates them at this ask if %s is placed (its grant names only placed "+
"modules' streams) and says asserted; grants_assert again shows what the bus holds then", missing, module)
return out, nil
}
// Answer is a tool's arguments read as text.
modules/mesh-grants/module.json
+1
View File
@@ -19,6 +19,7 @@
"nats.nats_users"
],
"emits": [
"asserted",
"membership-changed",
"user-list-changed"
],