diff --git a/modules/mesh-grants/cmd/mesh-grants/create.go b/modules/mesh-grants/cmd/mesh-grants/create.go new file mode 100644 index 0000000..c155393 --- /dev/null +++ b/modules/mesh-grants/cmd/mesh-grants/create.go @@ -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 +} diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 740e700..e5466c5 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -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 } diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index 51a662b..b4eac35 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -91,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 " + diff --git a/modules/mesh-grants/go.mod b/modules/mesh-grants/go.mod index b551d84..393b876 100644 --- a/modules/mesh-grants/go.mod +++ b/modules/mesh-grants/go.mod @@ -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 ) diff --git a/modules/mesh-grants/go.sum b/modules/mesh-grants/go.sum index 56f4d59..919fd4e 100644 --- a/modules/mesh-grants/go.sum +++ b/modules/mesh-grants/go.sum @@ -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= diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 17d4053..fe5cd07 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -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 { @@ -721,6 +751,95 @@ 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) @@ -813,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 diff --git a/modules/mesh-grants/internal/grants/assertions.go b/modules/mesh-grants/internal/grants/assertions.go index a2f1ee2..bac47eb 100644 --- a/modules/mesh-grants/internal/grants/assertions.go +++ b/modules/mesh-grants/internal/grants/assertions.go @@ -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 `: `. +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 diff --git a/modules/mesh-grants/internal/grants/follow.go b/modules/mesh-grants/internal/grants/follow.go index b8a33de..a3a6400 100644 --- a/modules/mesh-grants/internal/grants/follow.go +++ b/modules/mesh-grants/internal/grants/follow.go @@ -36,8 +36,9 @@ type Follower struct { watched map[string]bool refused map[string]string have map[string]string - // held is what the last pass held back, said again only when it changes. - held 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() { @@ -145,8 +146,9 @@ func (f *Follower) open() { } } -// said logs every membership the pass held back, each with what it would lose (issue 1945), when the set changes from -// the last pass's. +// 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 @@ -160,6 +162,15 @@ func (f *Follower) said(res Result) { } } } + 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. diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 4df21c5..2a480ed 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -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 ( diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index d1b5d3e..b78b602 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -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, ` at : created …`, and + // NotAsserted each creation refused, ` : ` (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 or drops only a verb the seat catalogue retired; any other 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) { diff --git a/modules/mesh-grants/internal/grants/tools.go b/modules/mesh-grants/internal/grants/tools.go index 16505da..7fe78c0 100644 --- a/modules/mesh-grants/internal/grants/tools.go +++ b/modules/mesh-grants/internal/grants/tools.go @@ -137,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) @@ -180,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") @@ -201,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. diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index 7af5032..97cbb27 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -19,6 +19,7 @@ "nats.nats_users" ], "emits": [ + "asserted", "membership-changed", "user-list-changed" ],