mesh-grants: create a declared bucket, log or stream the bus lacks, so a new module state no longer waits for a controller start
A module's new state reached the bus only when the controller started (issue 1790, met three times: puts, queues, mesh-vault's bus-accounts). The operator allowed one more grant in the controller's composer on 2026-10-26 08:54, creation only: the part's process now creates each bucket, log or stream a placed module's manifest declares and the bus lacks, with the same configuration the controller asserts (mesh-sdk BucketStreamConfig, Log.Config, ModuleStreamConfig), on the event that brought the manifest, never on a timer. It says asserted and logs one line per module; a stream the bus holds is never updated or deleted, its difference only said; a create refused is said and tried at the next event; one created at this manifest and gone since is made again only by hand. grants_assert asks the process to do the same for one module now. The user list, the grants and any change to a held stream stay in shadow. mesh-sdk moves to fix/1790-grants-create-buckets, whose composer grants this, so the shadow list is composed as the controller's. Issue: 1790
This commit is contained in:
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 " +
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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 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) {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -19,6 +19,7 @@
|
||||
"nats.nats_users"
|
||||
],
|
||||
"emits": [
|
||||
"asserted",
|
||||
"membership-changed",
|
||||
"user-list-changed"
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user